Linux网络协议栈：NAPI机制与处理流程分析（图解）

Linux网络协议栈：NAPI机制与处理流程分析（图解）_RToax-CSDN博客_netif_napi_addTable of ContentsNAPI机制NAPI缺陷使用 NAPI 先决条件非NAPI帧的接收netif_rx -将网卡中收到的数据包放到系统中的接收队列中enqueue_to_backlog____napi_schedule函数NAPI方式NAPI帧的接收NAPI接口struct napi_struct结构 -内核处理软中断的入口netif_napi_add函数 -驱动初始时向内核注册软软中断处理回调poll函数__napi_schedul...https://blog.csdn.net/Rong_Toa/article/details/109401935

随着网络带宽的发展，网速越来越快，之前的中断收包模式已经无法适应目前千兆，万兆的带宽了。如果每个数据包大小等于MTU大小1460字节。当驱动以千兆网速收包时，CPU将每秒被中断91829次。在以MTU收包的情况下都会出现每秒被中断10万次的情况。过多的中断会引起一个问题，CPU一直陷入硬中断而没有时间来处理别的事情了。为了解决这个问题，内核在2.6中引入了NAPI机制。

NAPI就是混合中断和轮询的方式来收包，当有中断来了，驱动关闭中断，通知内核收包，内核软中断轮询当前网卡，在规定时间尽可能多的收包。时间用尽或者没有数据可收，内核再次开启中断，准备下一次收包。

本文将介绍Linux内核中的NAPI：Linux网络设备驱动程序中的一种支持新一代网络适配器的架构。

NAPI机制
New API（NAPI）用于支持高速网卡处理网络数据包的一种机制 - 例如在Linux 2.6内核版本中引入的千兆以太网卡，后来又被移植到了2.4.x版本中。

NAPI 是 Linux 上采用的一种提高网络处理效率的技术，它的核心概念就是不采用中断的方式读取数据，而代之以首先采用中断唤醒数据接收的服务程序，然后 POLL 的方法来轮询数据。随着网络的接收速度的增加，NIC 触发的中断能做到不断减少，目前 NAPI 技术已经在网卡驱动层和网络层得到了广泛的应用，驱动层次上已经有 E1000 系列网卡，RTL8139 系列网卡，3c50X 系列等主流的网络适配器都采用了这个技术，而在网络层次上，NAPI 技术已经完全被应用到了著名的netif_rx 函数中间，并且提供了专门的 POLL 方法--process_backlog 来处理轮询的方法；根据实验数据表明采用NAPI技术可以大大改善短长度数据包接收的效率，减少中断触发的时间。

NAPI 对数据包到达的事件的处理采用轮询方法，在数据包达到的时候，NAPI 就会强制执行dev->poll方法。而和不像以前的驱动那样为了减少包到达时间的处理延迟，通常采用中断的方法来进行。

以前的网络设备驱动程序架构已经不能适用于每秒产生数千个中断的高速网络设备，并且它可能导致整个系统处于饥饿状态（译者注：饥饿状态的意思是系统忙于处理中断程序，没有时间执行其他程序）。有些网络设备具有中断合并，或者将多个数据包组合在一起来减少中断请求这种高级功能。

在内核没有使用NAPI来支持这些高级特性之前，这些功能只能全部在设备驱动程序中结合抢占机制（例如基于定时器中断），甚至中断程序范围之外的轮询程序（例如：内核线程，tasklet等）中实现。

正如我们看到的，网络子系统中加入的这个新特性是用于更好的支持中断缓解和数据包限制等功能，更重要的是它允许内核通过round-robin策略（轮询即Round Robin，一种负载均衡策略）将负载分发到不同网络设备上。

NAPI特性的添加不会影响内核的向后兼容性。

NAPI缺陷
NAPI 存在一些比较严重的缺陷：

1． 对于上层的应用程序而言，系统不能在每个数据包接收到的时候都可以及时地去处理它，而且随着传输速度增加，累计的数据包将会耗费大量的内存，经过实验表明在 Linux 平台上这个问题会比在 FreeBSD 上要严重一些；
2． 另外一个问题是对于大的数据包处理比较困难，原因是大的数据包传送到网络层上的时候耗费的时间比短数据包长很多（即使是采用 DMA 方式），所以正如前面所说的那样，NAPI 技术适用于对高速率的短长度数据包的处理。
 

使用 NAPI 先决条件
驱动可以继续使用老的 2.4 内核的网络驱动程序接口，NAPI 的加入并不会导致向前兼容性的丧失，但是 NAPI 的使用至少要得到下面的保证：

1. 设备需要有足够的缓冲区，保存多个数据分组。要使用 DMA 的环形输入队列（也就是 ring_dma，这个在 2.4 驱动中关于 Ethernet 的部分有详细的介绍），或者是有足够的内存空间缓存驱动获得的包。
2. 可以禁用当前设备中断，然而不影响其他的操作。在发送/接收数据包产生中断的时候有能力关断 NIC 中断的事件处理，并且在关断 NIC 以后，并不影响数据包接收到网络设备的环形缓冲区（以下简称 rx-ring）处理队列中。
当前大部分的设备都支持NAPI，但是为了对之前的保持兼容，内核还是对之前中断方式提供了兼容。我们先看下NAPI具体的处理方式。

我们都知道中断分为中断上半部和下半部，上半部完成的任务很是简单，仅仅负责把数据保存下来；而下半部负责具体的处理。为了处理下半部，每个CPU有维护一个softnet_data结构（下文将进行讲解）。我们不对此结构做详细介绍，仅仅描述和NAPI相关的部分。结构中有一个poll_list字段，连接所有的轮询设备。还 维护了两个队列input_pkt_queue和process_queue。这两个用户传统不支持NAPI方式的处理。前者由中断上半部的处理函数把数据包入队，在具体的处理时，使用后者做中转，相当于前者负责接收，后者负责处理。最后是一个napi_struct的backlog，代表一个虚拟设备供轮询使用。在支持NAPI的设备下，每个设备具备一个缓冲队列，存放到来数据。每个设备对应一个napi_struct结构，该结构代表该设备存放在poll_list中被轮询。而设备还需要提供一个poll函数，在设备被轮询到后，会调用poll函数对数据进行处理。基本逻辑就是这样，下文将给出具体流程。

/*
 * 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
}
 

非NAPI帧的接收
我们将讨论内核在接收一个数据帧后的大致处理流程，不会详细叙述所有细节。

我们认为有必要先了解一下传统的数据包处理流程以便更好的理解NAPI和传统收包方式的区别。

在传统的收包方式中（如下图）数据帧向网络协议栈中传递发生在中断上下文（在接收数据帧时）中调用netif_rx的函数中。 这个函数还有一个变体netif_rx_ni，他被用于中断上下文之外。

图 - Non-NAPI frame reception

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

netif_rx - 将网卡中收到的数据包放到系统中的接收队列中
int netif_rx(struct sk_buff *skb)
{
    int ret;
 
    /* if netpoll wants it, pretend we never saw it */
    /*如果是net_poll想要的，则不作处理*/
    if (netpoll_rx(skb))
        return NET_RX_DROP;
    /*检查时间戳*/
    net_timestamp_check(netdev_tstamp_prequeue, skb);
 
    trace_netif_rx(skb);
#ifdef CONFIG_RPS
    if (static_key_false(&rps_needed)) {
        struct rps_dev_flow voidflow, *rflow = &voidflow;
        int cpu;
        /*禁用抢占*/
        preempt_disable();
        rcu_read_lock();
        
        cpu = get_rps_cpu(skb->dev, skb, &rflow);
        if (cpu < 0)
            cpu = smp_processor_id();
        /*把数据入队*/
        ret = enqueue_to_backlog(skb, cpu, &rflow->last_qtail);
 
        rcu_read_unlock();
        preempt_enable();
    } else
#endif
    {
        unsigned int qtail;
        ret = enqueue_to_backlog(skb, get_cpu(), &qtail);
        put_cpu();
    }
    return ret;
}
 

enqueue_to_backlog

中间RPS暂时不关心，这里直接调用enqueue_to_backlog放入CPU的全局队列input_pkt_queue

static int enqueue_to_backlog(struct sk_buff *skb, int cpu,
                  unsigned int *qtail)
{
    struct softnet_data *sd;
    unsigned long flags;
    /*获取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) {
        /*如果input_pkt_queue不为空，说明虚拟设备已经得到调度，此时仅仅把数据加入
            input_pkt_queue队列即可
        */
        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
         */
         /*否则需要调度backlog 即虚拟设备，然后再入队。napi_struct结构中的state字段如果标记了NAPI_STATE_SCHED,则表明该设备已经在调度，不需要再次调度*/
        if (!__test_and_set_bit(NAPI_STATE_SCHED, &sd->backlog.state)) {
            if (!rps_ipi_queued(sd))
                ____napi_schedule(sd, &sd->backlog);
        }
        goto enqueue;
    }
    /*到这里缓冲区已经不足了，必须丢弃*/
    sd->dropped++;
    rps_unlock(sd);
    local_irq_restore(flags);
    atomic_long_inc(&skb->dev->rx_dropped);
    kfree_skb(skb);
    return NET_RX_DROP;
}
该函数逻辑也比较简单，主要注意的是设备必须先添加调度然后才能接受数据，添加调度调用了____napi_schedule函数，该函数把设备对应的napi_struct结构插入到softnet_data的poll_list链表尾部，然后唤醒软中断，这样在下次软中断得到处理时，中断下半部就会得到处理。不妨看下源码

____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);
}
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
}

如上所示，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队列中传递给了网络协议栈，现在数据包可以被处理了。

NAPI方式
NAPI的方式相对于非NAPI要简单许多，看下e100网卡的中断处理函数e100_intr，核心部分

if (likely(napi_schedule_prep(&nic->napi))) {
        e100_disable_irq(nic);//屏蔽当前中断
        __napi_schedule(&nic->napi);//把设备加入到轮训队列
    }
if条件检查当前设备是否 可被调度，主要检查两个方面：

1、是否已经在调度
2、是否禁止了napi pending.
如果符合条件，就关闭当前设备的中断，调用__napi_schedule函数把设备假如到轮训列表，从而开启轮询模式。

分析：结合上面两种方式，还是可以发现两种方式的异同。其中softnet_data作为主导结构，在NAPI的处理方式下，主要维护轮询链表。NAPI设备均对应一个napi_struct结构，添加到链表中；非NAPI没有对应的napi_struct结构，为了使用NAPI的处理流程，使用了softnet_data结构中的back_log作为一个虚拟设备添加到轮询链表。同时由于非NAPI设备没有各自的接收队列，所以利用了softnet_data结构的input_pkt_queue作为全局的接收队列。这样就处理而言，可以和NAPI的设备进行兼容。但是还有一个重要区别，在NAPI的方式下，首次数据包的接收使用中断的方式，而后续的数据包就会使用轮询处理了；而非NAPI每次都是通过中断通知。

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

图 - NAPI frame reception

因此网卡支持NAPI必须满足几个条件：

驱动程序不再使用数据包接收队列
网卡本身需要维护一个缓冲区来保存接收到数据包，并且可以禁止中断。
这种方法减少了中断的产生并且在突发情况下减少了丢包的可能性，避免了接收队列的饱和。

从NAPI实现的角度来看，与传统收包方式的不同地方在中断程序和轮询函数上（在net_device结构体中定义），定义如下：

int (*poll)(struct net_device *dev, int *budget);
除此之外，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函数的执行步骤如下：

回收当前处理器的poll_list链表的引用。
将jiffies的值保存在start_time变量中。
设置轮询的budget（预算，可处理的数据包数量）为netdev_budget变量的初始值（这个值可以通过 /proc/sys/net/core/netdev_budget来配置）
轮询poll_list链表中的每个设备，直到你的budget用完，当你的运行时间还没有超过一个jiffies时:
a) 如果quantum（配额）为正值则调用设备的poll()函数，否则将weight的值加到quantum中，将设备放回poll_list链表；
a.1) 如果poll()函数返回一个非零值，将weight的值设置到quantum中然后将设备放回poll_list链表；
a.2) 如果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，此时会导致一个轮询周期的时间变长。

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

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;
}

下图分别展示了非NAPI和NAPI模型中数据包接收处理过程的时序图：

图 - 非NAPI模型的时序图

图 - NAPI模型的时序图

NAPI接口
struct napi_struct结构  - 内核处理软中断的入口
struct napi_struct 是内核处理软中断的入口，每个net_device都对应一个napi_struct，驱动在硬中断中将自己的napi_struct挂载到CPU的收包队列softnet_data。内核在软中断中轮询该队列，并执行napi_sturct中的回调函数int(*poll)(struct napi_struct *, int);

在poll函数中，驱动将网卡数据转换成skb_buff形式，最终发往协议栈。也就是说，协议栈对数据包的处理，使用的是软中断的时间片。如果协议栈处理耗费了过多的CPU时间的化，会直接影响到设备的网络性能。

/*
 * 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; //每次轮询最大处理数据包数量
    unsigned int        gro_count;
    int         (*poll)(struct napi_struct *, int);//轮询设备的回调函数
#ifdef CONFIG_NETPOLL
    int         poll_owner;
#endif
    struct net_device   *dev;
    struct sk_buff      *gro_list;
    struct sk_buff      *skb;
    struct hrtimer      timer;
    struct list_head    dev_list;
    struct hlist_node   napi_hash_node;
    unsigned int        napi_id;
};

有了保存数据的结构体，让我们在看看为它配套提供的接口函数吧：

netif_napi_add函数 - 驱动初始时向内核注册软软中断处理回调poll函数
驱动在初始化net_device时通过这函数将通过这个函数绑定一个napi_struct结构。驱动需要在这里注册软中断中用于轮询的网卡的poll函数。

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);
    hrtimer_init(&napi->timer, CLOCK_MONOTONIC, HRTIMER_MODE_REL_PINNED);
    napi->timer.function = napi_watchdog;
    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_NETPOLL
    napi->poll_owner = -1;
#endif
    set_bit(NAPI_STATE_SCHED, &napi->state);
    napi_hash_add(napi);
}

__napi_schedule函数 - 网卡硬件中断用来触发软中断

__napi_schedule函数，为驱动硬件中断提供的接口，驱动在硬件中断中，将自己的napi_struct挂载到当前CPU的softnet_data上。

/**
 * __napi_schedule - schedule for receive
 * @n: entry to schedule
 *
 * The entry's receive function will be scheduled to run.
 * Consider using __napi_schedule_irqoff() if hard irqs are masked.
 */
void __napi_schedule(struct napi_struct *n)
{
    unsigned long flags;
 
    local_irq_save(flags);
    ____napi_schedule(this_cpu_ptr(&softnet_data), n);
    local_irq_restore(flags);
}
 
/* Called with irq disabled */
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); //设置了软中断接收标志位
}

napi_schedule_prep函数 - 对napi_struct进行检查
napi_schedule_prep函数是上面__napi_schedule的配套函数，用于__napi_schedule调用前对napi_struct进行检查。前面博文e1000网卡的中断函数就是这样调用的。

if (likely(napi_schedule_prep(&adapter->napi))) {
        adapter->total_tx_bytes = 0;
        adapter->total_tx_packets = 0;
        adapter->total_rx_bytes = 0;
        adapter->total_rx_packets = 0;
        __napi_schedule(&adapter->napi);
 }
判断NAPI是否可以调度。如果NAPI没有被禁止，且不存在已被调度的NAPI，则允许调度NAPI，因为同一时刻只允许有一个NAPI poll instance。测试napi.state字段，只有当其不是NAPI_STATE_SCHED时，返回真，并设置为NAPI_STATE_SCHED.

/**
 *  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.
 */
bool napi_schedule_prep(struct napi_struct *n)
{
    unsigned long val, new;
 
    do {
        val = READ_ONCE(n->state);
        if (unlikely(val & NAPIF_STATE_DISABLE))
            return false;
        new = val | NAPIF_STATE_SCHED;
 
        /* Sets STATE_MISSED bit if STATE_SCHED was already set
         * This was suggested by Alexander Duyck, as compiler
         * emits better code than :
         * if (val & NAPIF_STATE_SCHED)
         *     new |= NAPIF_STATE_MISSED;
         */
        new |= (val & NAPIF_STATE_SCHED) / NAPIF_STATE_SCHED *
                           NAPIF_STATE_MISSED;
    } while (cmpxchg(&n->state, val, new) != val);
 
    return !(val & NAPIF_STATE_SCHED);
}

上面的三个函数netif_napi_add,__napi_schedule,napi_schedule_prep是驱动使用NAPI收包机制的接口，下面再看看内核软中断使用NAPI的接口函数吧。

napi_poll函数 - 用于调用收包poll函数
这函数是被软中断处理函数net_rx_action调用的。这个函数将在napi_struct.weight规定的时间内，被net_rx_action循环调用，直到时间片用尽或者网卡当前DMA中所有缓存的数据包被处理完。如果是由于时间片用尽而退出的的话，napi_struct会重新挂载到softnet_data上，而如果是所有数据包处理完退出的，napi_struct会从softnet_data上移除并重新打开网卡硬件中断。

static int napi_poll(struct napi_struct *n, struct list_head *repoll)
{
    void *have;
    int work, weight;
 
    list_del_init(&n->poll_list);
 
    have = netpoll_poll_lock(n);
 
    weight = n->weight;
 
    /* This NAPI_STATE_SCHED test is for avoiding a race
     * with netpoll's poll_napi().  Only the entity 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;
    if (test_bit(NAPI_STATE_SCHED, &n->state)) {
        work = n->poll(n, weight);  //调用网卡注册的poll函数
        trace_napi_poll(n, work, weight);
    }
 
    WARN_ON_ONCE(work > weight);
 
    if (likely(work < weight))
        goto out_unlock;
 
    /* Drivers must not modify the NAPI state if they
     * consume the entire weight.  In such cases this code
     * still "owns" the NAPI instance and therefore can
     * move the instance around on the list at-will.
     */
    if (unlikely(napi_disable_pending(n))) {
        napi_complete(n);
        goto out_unlock;
    }
 
    if (n->gro_list) {
        /* flush too old packets
         * If HZ < 1000, flush all packets.
         */
        napi_gro_flush(n, HZ >= 1000);
    }
 
    /* Some drivers may have called napi_schedule
     * prior to exhausting their budget.
     */
    if (unlikely(!list_empty(&n->poll_list))) {
        pr_warn_once("%s: Budget exhausted after napi rescheduled\n",
                 n->dev ? n->dev->name : "backlog");
        goto out_unlock;
    }
 
    list_add_tail(&n->poll_list, repoll);
 
out_unlock:
    netpoll_poll_unlock(have);
 
    return work;
}

napi_gro_receive函数 - poll函数用来将网卡上的数据包发给协议栈处理

准确来说napi_gro_receive函数是驱动通过poll注册，内核调用的函数。通过这函数的的调用，skb将会传给协议栈的入口函数__netif_receive_skb。dev_gro_receive函数用于对数据包的合并，他将合并napi_struct.gro_list链表上的skb。GRO是一个网络子系统的另一套机制，以后再看。

gro_result_t napi_gro_receive(struct napi_struct *napi, struct sk_buff *skb)
{
    skb_mark_napi_id(skb, napi);
    trace_napi_gro_receive_entry(skb);
 
    skb_gro_reset_offset(skb);
 
    return napi_skb_finish(dev_gro_receive(napi, skb), skb);
}
 

总结
netif_napi_add：驱动初始时向内核注册软软中断处理回调poll函数
napi_schedule_prep函数是上面__napi_schedule的配套函数，用于__napi_schedule调用前对napi_struct进行检查
__napi_schedule：网卡硬件中断用来触发软中断
napi_poll：软中断处理函数net_rx_action用来回调上面驱动初始化是通过netif_napi_add注册的回调收包poll函数
napi_gro_receive：poll函数用来将网卡上的数据包发给协议栈处理。
到这，NAPI机制下的收包处理流程就很清晰了。

IRQ->__napi_schedule->进入软中断->net_rx_action->napi_poll->驱动注册的poll->napi_gro_receive。

IRQ
    ->__napi_schedule
        ->进入软中断
            ->net_rx_action
                ->napi_poll
                    ->驱动注册的poll
                        ->napi_gro_receive。

 

E1000网卡驱动程序对NAPI的支持
上面已经介绍过了，使用NAPI需要在编译内核的时候选择打开相应网卡设备的NAPI支持选项，对于E1000网卡来说就是CONFIG_E1000_NAPI宏。

e1000_probe函数 - E1000网卡的初始化函数
E1000网卡的初始化函数，也就是通常所说的probe方法，定义为e1000_probe（）：

static int __devinit e1000_probe(struct pci_dev *pdev, const struct pci_device_id *ent)
{
    struct net_device *netdev;
    struct e1000_adapter *adapter;
    static int cards_found = 0;
    unsigned long mmio_start;
    int mmio_len;
    int pci_using_dac;
    int i;
    int err;
    uint16_t eeprom_data;
 
    if((err = pci_enable_device(pdev)))
        return err;
    /*
    在这里设置PCI设备的DMA掩码，如果这个设备支持DMA传输，则掩码置位。
    */
    if(!(err = pci_set_dma_mask(pdev, PCI_DMA_64BIT))) {
        pci_using_dac = 1;
    } else {
        if((err = pci_set_dma_mask(pdev, PCI_DMA_32BIT))) {
            E1000_ERR("No usable DMA configuration, aborting\n");
            return err;
        }
        pci_using_dac = 0;
    }
 
    if((err = pci_request_regions(pdev, e1000_driver_name)))
        return err;
 
    pci_set_master(pdev);
    /*
    为e1000网卡对应的net_device结构分配内存。
    */
    netdev = alloc_etherdev(sizeof(struct e1000_adapter));
    if(!netdev) {
        err = -ENOMEM;
        goto err_alloc_etherdev;
    }
 
    SET_MODULE_OWNER(netdev);
 
    pci_set_drvdata(pdev, netdev);
    adapter = netdev->priv;
    adapter->netdev = netdev;
    adapter->pdev = pdev;
    adapter->hw.back = adapter;
 
    mmio_start = pci_resource_start(pdev, BAR_0);
    mmio_len = pci_resource_len(pdev, BAR_0);
 
    adapter->hw.hw_addr = ioremap(mmio_start, mmio_len);
    if(!adapter->hw.hw_addr) {
        err = -EIO;
        goto err_ioremap;
    }
 
    for(i = BAR_1; i <= BAR_5; i++) {
        if(pci_resource_len(pdev, i) == 0)
            continue;
        if(pci_resource_flags(pdev, i) & IORESOURCE_IO) {
            adapter->hw.io_base = pci_resource_start(pdev, i);
            break;
        }
    }
 
    /*
    将e1000网卡驱动程序的相应函数注册到net_device结构的成员函数上。这里值得注意的
    是如果定义了设备的CONFIG_E1000_NAPI宏，则设备对应的poll方法被注册为e1000_clean。
    在网络设备初始化时（net_dev_init()函数）将所有的设备的poll方法注册为系统默认
    函数process_backlog（），该函数的处理方法就是从CPU相关队列softnet_data的输入
    数据包队列中读取skb，然后调用netif_receive_skb（）函数提交给上层协议继续处
    理。设备的poll方法是在软中断处理函数中调用的。
    */
    netdev->open = &e1000_open;
    netdev->stop = &e1000_close;
    netdev->hard_start_xmit = &e1000_xmit_frame;
    netdev->get_stats = &e1000_get_stats;
    netdev->set_multicast_list = &e1000_set_multi;
    netdev->set_mac_address = &e1000_set_mac;
    netdev->change_mtu = &e1000_change_mtu;
    netdev->do_ioctl = &e1000_ioctl;
    netdev->tx_timeout = &e1000_tx_timeout;
    netdev->watchdog_timeo = 5 * HZ;
#ifdef CONFIG_E1000_NAPI
    netdev->poll = &e1000_clean;
    netdev->weight = 64;
#endif
    netdev->vlan_rx_register = e1000_vlan_rx_register;
    netdev->vlan_rx_add_vid = e1000_vlan_rx_add_vid;
    netdev->vlan_rx_kill_vid = e1000_vlan_rx_kill_vid;
    /*
    这些就是利用ifconfig能够看到的内存起始地址，以及基地址。
    */
    netdev->irq = pdev->irq;
    netdev->mem_start = mmio_start;
    netdev->mem_end = mmio_start + mmio_len;
    netdev->base_addr = adapter->hw.io_base;
 
    adapter->bd_number = cards_found;
 
    if(pci_using_dac)
        netdev->features |= NETIF_F_HIGHDMA;
 
    /* MAC地址是存放在网卡设备的EEPROM上的，现在将其拷贝出来。 */
    e1000_read_mac_addr(&adapter->hw);
    memcpy(netdev->dev_addr, adapter->hw.mac_addr, netdev->addr_len);
    if(!is_valid_ether_addr(netdev->dev_addr)) {
        err = -EIO;
        goto err_eeprom;
    }
    /*
    这里初始化三个定时器列表，以后对内核Timer的实现进行分析，这里就不介绍了。
    */
    init_timer(&adapter->tx_fifo_stall_timer);
    adapter->tx_fifo_stall_timer.function = &e1000_82547_tx_fifo_stall;
    adapter->tx_fifo_stall_timer.data = (unsigned long) adapter;
 
    init_timer(&adapter->watchdog_timer);
    adapter->watchdog_timer.function = &e1000_watchdog;
    adapter->watchdog_timer.data = (unsigned long) adapter;
 
    init_timer(&adapter->phy_info_timer);
    adapter->phy_info_timer.function = &e1000_update_phy_info;
    adapter->phy_info_timer.data = (unsigned long) adapter;
 
    INIT_TQUEUE(&adapter->tx_timeout_task,
    (void (*)(void *))e1000_tx_timeout_task, netdev);
    /*
    这里调用网络设备注册函数将当前网络设备注册到系统的dev_base[]设备数组当中，
    并且调用设备的probe函数，对于以太网来说，就是ethif_probe（）函数。相关的说明
    见内核网络设备操作部分的分析。
    调用关系：register_netdev （）->register_netdevice（）
    */
    register_netdev(netdev);
 
    netif_carrier_off(netdev);
    netif_stop_queue(netdev);
 
    e1000_check_options(adapter);
}

e1000_open函数 - E1000网卡驱动程序的open方法
在分析网卡接收数据包的过程中，设备的open方法是值得注意的，因为在这里对网卡设备的各种数据结构进行了初始化，特别是环形缓冲区队列。E1000网卡驱动程序的open方法注册为e1000_open（）：

static int e1000_open(struct net_device *netdev)
{
    struct e1000_adapter *adapter = netdev->priv;
    int err;
 
    /* allocate transmit descriptors */
 
    if((err = e1000_setup_tx_resources(adapter)))
        goto err_setup_tx;
 
    /* allocate receive descriptors */
 
    if((err = e1000_setup_rx_resources(adapter)))
        goto err_setup_rx;
 
    if((err = e1000_up(adapter)))
        goto err_up;
}

事实上e1000_open（） 函数调用了e1000_setup_rx_resources（）函数为其环形缓冲区分配资源。

e1000设备的接收方式是一种缓冲方式，能显著的降低 CPU接收数据造成的花费，接收数据之前，软件需要预先分配一个 DMA 缓冲区，一般对于传输而言，缓冲区最大为 8Kbyte 并且把物理地址链接在描述符的 DMA 地址描述单元，另外还有两个双字的单元表示对应的 DMA 缓冲区的接收状态。

e1000_setup_rx_resources函数 - 为其环形缓冲区分配资源
在 /driver/net/e1000/e1000/e1000.h 中对于环形缓冲队列描述符的数据单元如下表示：

struct e1000_desc_ring {
    void *desc; /* 指向描述符环状缓冲区的指针。*/
    dma_addr_t dma; /* 描述符环状缓冲区物理地址，也就是DMA缓冲区地址*/
    unsigned int size; /* 描述符环状缓冲区的长度（用字节表示）*/
    unsigned int count; /* 缓冲区内描述符的数量，这个是系统初始化时规定好的，它
    决定该环形缓冲区有多少描述符（或者说缓冲区）可用*/
    unsigned int next_to_use; /* 下一个要使用的描述符。*/
    unsigned int next_to_clean; /* 下一个待删除描述符。*/
    struct e1000_buffer *buffer_info; /* 缓冲区信息结构数组。*/
};
static int e1000_setup_rx_resources(struct e1000_adapter *adapter)
{
    /*将环形缓冲区取下来*/
    struct e1000_desc_ring *rxdr = &adapter->rx_ring;
    struct pci_dev *pdev = adapter->pdev;
    int size;
 
    size = sizeof(struct e1000_buffer) * rxdr->count;
    /*
    为每一个描述符缓冲区分配内存，缓冲区的数量由count决定。
    */
    rxdr->buffer_info = kmalloc(size, GFP_KERNEL);
    if(!rxdr->buffer_info) {
        return -ENOMEM;
    }
    memset(rxdr->buffer_info, 0, size);
 
    /* Round up to nearest 4K */
 
    rxdr->size = rxdr->count * sizeof(struct e1000_rx_desc);
    E1000_ROUNDUP(rxdr->size, 4096);
    /*
    调用pci_alloc_consistent（）函数为系统分配DMA缓冲区。
    */
    rxdr->desc = pci_alloc_consistent(pdev, rxdr->size, &rxdr->dma);
 
    if(!rxdr->desc) {
        kfree(rxdr->buffer_info);
        return -ENOMEM;
    }
    memset(rxdr->desc, 0, rxdr->size);
 
    rxdr->next_to_clean = 0;
    rxdr->next_to_use = 0;
 
    return 0;
}

request_irq函数 - 向系统申请irq中断号
在e1000_up（）函数中，调用request_irq（）向系统申请irq中断号，然后将e1000_intr（）中断处理函数注册到系统当中，系统有一个中断向量表irq_desc[]。然后使能网卡的中断。

接 下来就是网卡处于响应中断的模式，这里重要的函数是 e1000_intr（）中断处理函数，关于这个函数的说明在内核网络设备操作笔记当中，这里就不重复了，但是重点强调的是中断处理函数中对NAPI部分 的处理方法，因此还是将该函数的源码列出，不过省略了与NAPI无关的处理过程：

static irqreturn_t e1000_intr(int irq, void *data, struct pt_regs *regs)
{
    struct net_device *netdev = data;
    struct e1000_adapter *adapter = netdev->priv;
    uint32_t icr = E1000_READ_REG(&adapter->hw, ICR);
    #ifndef CONFIG_E1000_NAPI
    unsigned int i;
    #endif
 
    if(!icr)
        return IRQ_NONE; /* Not our interrupt */
 
#ifdef CONFIG_E1000_NAPI
    /*
    如果定义了采用NAPI模式接收数据包，则进入这个调用点。
    首先调用netif_rx_schedule_prep(dev)，确定设备处于运行，而且设备还没有被添加
    到网络层的 POLL 处理队列中，在调用 netif_rx_schedule之前会调用这个函数。
    接下来调用 __netif_rx_schedule(dev)，将设备的 POLL 方法添加到网络层次的
     POLL 处理队列中去，排队并且准备接收数据包，在使用之前需要调用
     netif_rx_reschedule_prep，并且返回的数为 1，并且触发一个 NET_RX_SOFTIRQ 的
    软中断通知网络层接收数据包。
    处理完成。
    */
    if(netif_rx_schedule_prep(netdev)) {
        /* Disable interrupts and register for poll. The flush
        of the posted write is intentionally left out.
        */
 
        atomic_inc(&adapter->irq_sem);
        E1000_WRITE_REG(&adapter->hw, IMC, ~0);
        __netif_rx_schedule(netdev);
    }
#else
    /*
    在中断模式下，就会调用net_if（）函数将数据包插入接收队列中，等待软中断处理。
    */
    for(i = 0; i < E1000_MAX_INTR; i++)
        if(!e1000_clean_rx_irq(adapter) &!e1000_clean_tx_irq(adapter))
            break;
#endif
 
    return IRQ_HANDLED;
}

__netif_rx_schedule函数 - 将设备POLL方法添到网络层次POLL处理队列
将设备的 POLL 方法添加到网络层次的 POLL 处理队列中去，排队并且准备接收数据包

下面介绍一下__netif_rx_schedule(netdev)函数的作用：

static inline void __netif_rx_schedule(struct net_device *dev)
{
    unsigned long flags;
    /* 获取当前CPU。 */
    int cpu = smp_processor_id();
 
    local_irq_save(flags);
    dev_hold(dev);
    
    /*将当前设备加入CPU相关全局队列softnet_data的轮询设备列表中，不过值得注意的
    是，这个列表中的设备不一定都执行轮询接收数据包，这里的poll_list只是表示当前设
    备需要接收数据，具体采用中断还是轮询的方式，取决于设备提供的poll方法。*/
    list_add_tail(&dev->poll_list, &softnet_data[cpu].poll_list);
    if (dev->quota < 0)
        /*对于e1000网卡的轮询机制，weight(是权，负担的意思)这个参数是64。而quota的意
        思是配额，限额。这两个参数在随后的轮询代码中出现频繁。*/
        dev->quota += dev->weight;
    else
        dev->quota = dev->weight;
    /*
    调用函数产生网络接收软中断。也就是系统将运行net_rx_action（）处理网络数据。
    */
    __cpu_raise_softirq(cpu, NET_RX_SOFTIRQ);
    local_irq_restore(flags);
}

net_rx_action函数 - 软中断处理函数
下半部：

下半部的处理函数，之前提到，网络数据包的接发对应两个不同的软中断，接收软中断NET_RX_SOFTIRQ的处理函数对应net_rx_action。

在内核网络设备操作阅读笔记当中已经介绍过net_rx_action（）这个重要的网络接收软中断处理函数了，不过这里为了清楚的分析轮询机制，需要再次分析这段代码：

static void net_rx_action(struct softirq_action *h)
{
    int this_cpu = smp_processor_id();
    /*获取当前CPU的接收数据队列。*/
    struct softnet_data *queue = &softnet_data[this_cpu];
    unsigned long start_time = jiffies;
    /*呵呵，这里先做个预算，限定我们只能处理这么多数据（300个）。*/
    int budget = netdev_max_backlog;
 
    br_read_lock(BR_NETPROTO_LOCK);
    local_irq_disable();
    /*
    进入一个循环，因为软中断处理函数与硬件中断并不是同步的，因此，我们此时并不知道
    数据包属于哪个设备，因此只能采取逐个查询的方式，遍历整个接收设备列表。
    */
    while (!list_empty(&queue->poll_list)) {
        struct net_device *dev;
        /*如果花费超过预算，或者处理时间超过1秒，立刻从软中断处理函数跳出，我想
        这可能是系统考虑效率和实时性，一次不能做过多的工作或者浪费过多的时间。*/
        if (budget <= 0 || jiffies - start_time > 1)
            goto softnet_break;
 
        local_irq_enable();
        /*从当前列表中取出一个接收设备。并根据其配额判断是否能够继续接收数据，如
        果配额不足（<=0），则立刻将该设备从设备列表中删除。并且再次插入队列当
        中，同时为该设备分配一定的配额，允许它继续处理数据包。
        如果此时配额足够，则调用设备的 poll方法，对于e1000网卡来说，如果采用中
        断方式处理数据，则调用系统默认poll方法process_backlog（），而对于采用
        NAPI 来说，则是调用e1000_clean（）函数了。记住这里第一次传递的预算是300
         ^_^。*/
        dev = list_entry(queue->poll_list.next, struct net_device, poll_list);
 
        if (dev->quota <= 0 || dev->poll(dev, &budget)) {
            local_irq_disable();
            list_del(&dev->poll_list);
            list_add_tail(&dev->poll_list, &queue->poll_list);
            if (dev->quota < 0)
                dev->quota += dev->weight;
            else
                dev->quota = dev->weight;
        } else {
            dev_put(dev);
            local_irq_disable();
        }
    }
 
    local_irq_enable();
    br_read_unlock(BR_NETPROTO_LOCK);
    return;
}

这里有处理方式比较直观，直接遍历poll_list链表，处理之前设置了两个限制：budget和time_limit。前者限制本次处理数据包的总量，后者限制本次处理总时间。只有二者均有剩余的情况下，才会继续处理。处理期间同样是开中断的，每次总是从链表表头取设备进行处理，如果设备被调度，其实就是检查NAPI_STATE_SCHED位，则调用 napi_struct的poll函数，处理结束如果没有处理完，则把设备移动到链表尾部，否则从链表删除。NAPI设备对应的poll函数会同样会调用__netif_receive_skb函数上传协议栈，这里就不做分析了，感兴趣可以参考e100的poll函数e100_poll。

e100_poll函数 - 非NAPI对应poll函数为process_backlog

而非NAPI对应poll函数为process_backlog。

static int process_backlog(struct napi_struct *napi, int quota)
{
    int work = 0;
    struct softnet_data *sd = container_of(napi, struct softnet_data, backlog);
 
#ifdef CONFIG_RPS
    /* Check if we have pending ipi, its better to send them now,
     * not waiting net_rx_action() end.
     */
    if (sd->rps_ipi_list) {
        local_irq_disable();
        net_rps_action_and_irq_enable(sd);
    }
#endif
    napi->weight = weight_p;
    local_irq_disable();
    while (work < quota) {
        struct sk_buff *skb;
        unsigned int qlen;
        /*涉及到两个队列process_queue和input_pkt_queue，数据包到来时首先填充input_pkt_queue，
        而在处理时从process_queue中取，根据这个逻辑，首次处理process_queue必定为空，检查input_pkt_queue
        如果input_pkt_queue不为空，则把其中的数据包迁移到process_queue中，然后继续处理，减少锁冲突。
        */
        while ((skb = __skb_dequeue(&sd->process_queue))) {
            local_irq_enable();
            /*进入协议栈*/
            __netif_receive_skb(skb);
            local_irq_disable();
            input_queue_head_incr(sd);
            if (++work >= quota) {
                local_irq_enable();
                return work;
            }
        }
 
        rps_lock(sd);
        qlen = skb_queue_len(&sd->input_pkt_queue);
        if (qlen)
            skb_queue_splice_tail_init(&sd->input_pkt_queue,
                           &sd->process_queue);
 
        if (qlen < quota - work) {
            /*
             * Inline a custom version of __napi_complete().
             * only current cpu owns and manipulates this napi,
             * and NAPI_STATE_SCHED is the only possible flag set on backlog.
             * we can use a plain write instead of clear_bit(),
             * and we dont need an smp_mb() memory barrier.
             */
            list_del(&napi->poll_list);
            napi->state = 0;
 
            quota = work + qlen;
        }
        rps_unlock(sd);
    }
    local_irq_enable();
 
    return work;
}

函数还是比较简单的，需要注意的每次处理都携带一个配额，即本次只能处理quota个数据包，如果超额了，即使没处理完也要返回，这是为了保证处理器的公平使用。处理在一个while循环中完成，循环条件正是work < quota，首先会从process_queue中取出skb,调用__netif_receive_skb上传给协议栈，然后增加work。当work即将大于quota时，即++work >= quota时，就要返回。当work还有剩余额度，但是process_queue中数据处理完了，就需要检查input_pkt_queue，因为在具体处理期间是开中断的，那么期间就有可能有新的数据包到来，如果input_pkt_queue不为空，则调用skb_queue_splice_tail_init函数把数据包迁移到process_queue。如果剩余额度足够处理完这些数据包，那么就把虚拟设备移除轮询队列。这里有些疑惑就是最后为何要增加额度，剩下的额度已经足够处理这些数据了呀？根据此流程不难发现，其实执行的是在两个队列之间移动数据包，然后再做处理。

e1000_clean函数 - 
下面介绍一下e1000网卡的轮询poll处理函数e1000_clean（），这个函数只有定义了NAPI宏的情况下才有效：

#ifdef CONFIG_E1000_NAPI
static int e1000_clean(struct net_device *netdev, int *budget)
{
 
    struct e1000_adapter *adapter = netdev->priv;
    /*计算一下我们要做的工作量，取系统给定预算（300）和我们网卡设备的配额之间的最
    小值，这样做同样是为了效率和实时性考虑，不能让一个设备在接收设备上占用太多的资
    源和时间。*/
    int work_to_do = min(*budget, netdev->quota);
    int work_done = 0;
    /*处理网卡向外发送的数据，这里我们暂时不讨论。*/
    e1000_clean_tx_irq(adapter);
    /*处理网卡中断收到的数据包，下面详细讨论这个函数的处理方法。*/
    e1000_clean_rx_irq(adapter, &work_done, work_to_do);
    /*从预算中减掉我们已经完成的任务，预算在被我们支出，^_^。同时设备的配额也不断
    的削减。*/
    *budget -= work_done;
    netdev->quota -= work_done;
    /* 如果函数返回时，完成的工作没有达到预期的数量，表明接收的数据包并不多，很快
    就全部处理完成了，我们就彻底完成了这次轮询任务，调用 netif_rx_complete（），
    把当前指定的设备从 POLL 队列中清除（注意如果在 POLL 队列处于工作状态的时候是
    不能把指定设备清除的，否则将会出错），然后使能网卡中断。*/
    if(work_done < work_to_do) {
        netif_rx_complete(netdev);
        e1000_irq_enable(adapter);
    }
    /*如果完成的工作大于预期要完成的工作，则表明存在问题，返回1，否则正常返回0。*/
 
    return (work_done >= work_to_do);
}

e1000_clean_rx_irq函数 - 处理网卡中断收到的数据包
设备轮询接收机制中最重要的函数就是下面这个函数，当然它同时也可以为中断接收机制所用，只不过处理过程有一定的差别。

上面看到budget -= napi_poll(n, &repoll);他会去调用我们驱动初始化时注册的poll函数，在e1000网卡中就是e1000_clean函数。

static boolean_t
#ifdef CONFIG_E1000_NAPI
e1000_clean_rx_irq(struct e1000_adapter *adapter, int *work_done,
int work_to_do)
#else
e1000_clean_rx_irq(struct e1000_adapter *adapter)
#endif
{
    /*这里很清楚，获取设备的环形缓冲区指针。*/
    struct e1000_desc_ring *rx_ring = &adapter->rx_ring;
    struct net_device *netdev = adapter->netdev;
    struct pci_dev *pdev = adapter->pdev;
    struct e1000_rx_desc *rx_desc;
    struct e1000_buffer *buffer_info;
    struct sk_buff *skb;
    unsigned long flags;
    uint32_t length;
    uint8_t last_byte;
    unsigned int i;
    boolean_t cleaned = FALSE;
    /*把i置为下一个要清除的描述符索引，因为在环形缓冲区队列当中，我们即使已经处理
    完一个缓冲区描述符，也不是将其删除，而是标记为已经处理，这样如果有新的数据需要
    使用缓冲区，只是将已经处理的缓冲区覆盖而已。*/
    i = rx_ring->next_to_clean;
    rx_desc = E1000_RX_DESC(*rx_ring, i);
    /*如果i对应的描述符状态是已经删除，则将这个缓冲区取出来给新的数据使用*/
    while(rx_desc->status & E1000_RXD_STAT_DD) {
          buffer_info = &rx_ring->buffer_info[i];
 
#ifdef CONFIG_E1000_NAPI
        /*在配置了NAPI的情况下，判断是否已经完成的工作？，因为是轮询机制，所以我
        们必须自己计算我们已经处理了多少数据。*/
        if(*work_done >= work_to_do)
            break;
 
        (*work_done)++;
#endif
 
        cleaned = TRUE;
        /*这个是DMA函数，目的是解除与DMA缓冲区的映射关系，这样我们就可以访问这个
        缓冲区，获取通过DMA传输过来的数据包（skb）。驱动程序在分配环形缓冲区的时
        候就将缓冲区与DMA进行了映射。*/
        pci_unmap_single(pdev,
        buffer_info->dma,
        buffer_info->length,
        PCI_DMA_FROMDEVICE);
 
        skb = buffer_info->skb;
        length = le16_to_cpu(rx_desc->length);
        /*对接收的数据包检查一下正确性。确认是一个正确的数据包以后，将skb的数据
        指针进行偏移。*/
        skb_put(skb, length - ETHERNET_FCS_SIZE);
 
        /* Receive Checksum Offload */
        e1000_rx_checksum(adapter, rx_desc, skb);
        /*获取skb的上层协议类型。这里指的是IP层的协议类型。*/
        skb->protocol = eth_type_trans(skb, netdev);
        #ifdef CONFIG_E1000_NAPI
        /*调用函数直接将skb向上层协议处理函数递交，而不是插入什么队列等待继续处
        理，因此这里可能存在一个问题，如果数据包比较大，处理时间相对较长，则可能
        造成系统效率的下降。*/
        netif_receive_skb(skb);
 
        #else /* CONFIG_E1000_NAPI */
        /*如果采用中断模式，则调用netif_rx（）将数据包插入队列中，在随后的软中
        断处理函数中调用netif_receive_skb（skb）向上层协议处理函数递交。这里就
        体现出了中断处理机制和轮询机制之间的差别。*/
        netif_rx(skb);
        #endif /* CONFIG_E1000_NAPI */
        /*用全局时间变量修正当前设备的最后数据包接收时间。*/
        netdev->last_rx = jiffies;
 
        rx_desc->status = 0;
        buffer_info->skb = NULL;
        /*这里是处理环形缓冲区达到队列末尾的情况，因为是环形的，所以到达末尾的下
        一个就是队列头，这样整个环形队列就不断的循环处理。然后获取下一个描述符的
        状态，看看是不是处于删除状态。如果处于这种状态就会将新到达的数据覆盖旧的
        的缓冲区，如果不处于这种状态跳出循环。并且将当前缓冲区索引号置为下一次查
        询的目标。*/
        if(++i == rx_ring->count) i = 0;
 
        rx_desc = E1000_RX_DESC(*rx_ring, i);
    }
 
    rx_ring->next_to_clean = i;
    /*为下一次接收skb做好准备，分配sk_buff内存。出于效率的考虑，如果下一个要使用
    的缓冲区的sk_buff还没有分配，就分配，如果已经分配，则可以重用。*/
    e1000_alloc_rx_buffers(adapter);
 
    return cleaned;
}

e1000_alloc_rx_buffers函数 - 分配环形接收缓冲区
下面分析的这个函数有助于我们了解环形接收缓冲区的结构和工作原理：

static void e1000_alloc_rx_buffers(struct e1000_adapter *adapter)
{
    struct e1000_desc_ring *rx_ring = &adapter->rx_ring;
    struct net_device *netdev = adapter->netdev;
    struct pci_dev *pdev = adapter->pdev;
    struct e1000_rx_desc *rx_desc;
    struct e1000_buffer *buffer_info;
    struct sk_buff *skb;
    int reserve_len = 2;
    unsigned int i;
    
    /*接收队列中下一个用到的缓冲区索引，初始化是0。并且获取该索引对应的缓冲区信息
    结构指针buffer_info。*/
    i = rx_ring->next_to_use;
    buffer_info = &rx_ring->buffer_info[i];
    
    /*如果该缓冲区还没有为sk_buff分配内存，则调用dev_alloc_skb函数分配内存，默认
    的e1000网卡的接收缓冲区长度是2048字节加上保留长度。
    注意：在e1000_open（）->e1000_up（）中已经调用了这个函数为环形缓冲区队列中的
    每一个缓冲区分配了sk_buff内存，但是如果接收到数据以后，调用netif_receive_skb
     (skb)向上层提交数据以后，这段内存将始终被这个skb占用（直到上层处理完以后才会
    调用__kfree_skb释放，但已经跟这里没有关系了），换句话说，就是当前缓冲区必须重
    新申请分配sk_buff内存，为了下一个数据做准备。*/
    while(!buffer_info->skb) {
        rx_desc = E1000_RX_DESC(*rx_ring, i)
        skb = dev_alloc_skb(adapter->rx_buffer_len + reserve_len);
 
        if(!skb) {
        /* Better luck next round */
            break;
        }
        skb_reserve(skb, reserve_len);
 
        skb->dev = netdev;
        /*映射DMA缓冲区，DMA通道直接将收到的数据写到我们提供的这个缓冲区内，每次
        必须将缓冲区与DMA通道解除映射关系，才能读取缓冲区内容。*/
        buffer_info->skb = skb;
        buffer_info->length = adapter->rx_buffer_len;
        buffer_info->dma = pci_map_single(pdev,
                                            skb->data,
                                            adapter->rx_buffer_len,
                                            PCI_DMA_FROMDEVICE);
 
        rx_desc->buffer_addr = cpu_to_le64(buffer_info->dma);
 
        if(++i == rx_ring->count) i = 0;
        buffer_info = &rx_ring->buffer_info[i];
    }
    rx_ring->next_to_use = i;
}

参考文章
《NAPI机制》- 简书

《NAPI(New API)的一些浅见》- 简书

《Linux协议栈--NAPI机制》 - CXD

《Linux NAPI处理流程分析》- CNBlogs

《网卡收包》- 简书

《Linux NAPI-compliant network device driver》

《Linux网络协议栈：网卡收包分析》

推荐阅读
《深入理解 Cilium 的 eBPF（XDP）收发包路径：数据包在Linux网络协议栈中的路径》

《iptables详解（1）：iptables概念》

《iptables详解（2）：路由表》

《eBPF.io eBPF文档：扩展的数据包过滤器（BPF）》

《Linux eBPF和XDP高速处理数据包；使用EBPF编写XDP网络过滤器；高性能ACL》

《深入理解 Cilium 的 eBPF 收发包路径》

《Cilium提供并透明地保护应用程序工作负载之间的网络连接和负载平衡：什么是eBPF和XDP？》

《DPDK 网卡收包流程》

《ethtool 原理介绍和解决网卡丢包排查思路》
————————————————
版权声明：本文为CSDN博主「rtoax」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/Rong_Toa/article/details/109401935