linux的NAPI方法收包

NAPI和传统收包方法的区别是：NAPI可以进一次中断收很多次的包，但是传统方法进一次中断后将包放到local cpu的softnet_data的input_queue，之后就退出中断。

一、传统方法
        进入处理程序后，首先从物理设备中将数据分组拷贝到内存中，然后分组是否合法，之后分配一个skb组包，然后调用netif_rx(skb)将包放入到softnet_data的input_queue中。在下图所示的传统API收包过程中，收包的IRQ是不需要被禁用的。因为将包放入到cpu的等待队列不会耗时太长。这也从另外一个角度说明，传统API只能适用与低速设备。因此每进一次中断，只收一个包。
        以3c501.c为例：irqreturn_t el_interrupt(int irq, void *dev_id) -> el_receive(dev) -> dev_alloc_skb(pkt_len+2); netif_rx(skb);
        netif_rx函数不是特定与网络设备的，真实物理的驱动可以调用它，一些虚拟接口，如ppp接口在处理完本层的业务后，剥离ppp头部，调用该函数处理包，之后在netif_receive_skb中会进入到IP层处理该数据报文。

int netif_rx(struct sk_buff *skb)
{
struct softnet_data *queue;
unsigned long flags;
...
local_irq_save(flags);
queue = &__get_cpu_var(softnet_data);
__get_cpu_var(netdev_rx_stat).total++;
if (queue->input_pkt_queue.qlen <= netdev_max_backlog) {  
  if (queue->input_pkt_queue.qlen) {  
enqueue:
  __skb_queue_tail(&queue->input_pkt_queue, skb);
  local_irq_restore(flags);
  return NET_RX_SUCCESS;
  }
napi_schedule(&queue->backlog);
  goto enqueue;
}

__get_cpu_var(netdev_rx_stat).dropped++;
local_irq_restore(flags);
kfree_skb(skb);
return NET_RX_DROP;
}
        netif_rx有两个返回值：NET_RX_SUCCESS和NET_RX_DROP。input_pkt_queue的类型是sk_buff_head。注意到该数据结构并没有使用内核常用的list_head作为保存skb的链表。softnet_data使用该数据结构存储与收包队列相关的元素，如skb、队列大小、lock等。如果使用list_head，那么必然要将qlen和lock放到softnet_data中，不够整洁。所以如果qlen比netdev_max_backlog还要大的时候，就会直接丢弃该包。netdev_max_backlog的值默认为1000，可以在/proc/sys/net/core/ netdev_max_backlog中设置。

struct sk_buff_head {
struct sk_buff *next;   /* These two members must be first */
struct sk_buff *prev;
__u32  qlen;
spinlock_t lock;
};
        如果sd queue的qlen为0，那么可能queue->backlog已经从当前cpu sd的poll_list移除，因此需要调用napi_schedule(&queue->backlog)重新将queue加入到相应的poll_list上。然后再__skb_queue_tail将skb入队列。
        input_pkt_queue中的skb何时被取出呢？当我们在netif_rx中调用napi_schedule(&queue->backlog)的时候，它首先检查这个napi_struct实例是否可用。然后调用__napi_schedule来做实际的调度工作。

void __napi_schedule(struct napi_struct *n)
{
unsigned long flags;
local_irq_save(flags);
list_add_tail(&n->poll_list, &__get_cpu_var(softnet_data).poll_list);
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
local_irq_restore(flags);
}
        本地CPU的softnet_data中，有一个poll_list字段，用于保存所有待调度的napi_struct实例。在上面代码中，将&queue->backlog加入这个调度队列上。之后引发软中断NET_RX_SOFTIRQ。该软中断的处理函数为net_rx_action。可以想见，在该函数中必然会遍历当前softnet_data上的所有napi_struct，包括传统收包API  netif_rx对应的napi_struct实例。

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;
  INIT_LIST_HEAD(&queue->poll_list);
queue->backlog.poll = process_backlog;
  queue->backlog.weight = weight_p;
...
}
        在net_dev_init中，初始化所有CPU上的softnet_data实例。包括初始化napi_struct实例的input_pkt_queue、poll_list，以便这个napi_struct能够在napi_schedule中挂载到softnet_data->poll_list下被调度。另外，设置这个napi_struct实例的weight为weight_p，一般地这个值64，也可以通过/proc/sys/net/core/dev_weight修改它的值。
        下面是process_backlog的实现。通过这段代码，也可以了解NAPI的poll函数的实现方式。其他设备napi实例的poll函数与下面这个大同小异。

static int process_backlog(struct napi_struct *napi, int quota)
{
int work = 0;
struct softnet_data *queue = &__get_cpu_var(softnet_data);
unsigned long start_time = jiffies;
napi->weight = weight_p;

do {
  struct sk_buff *skb;
  local_irq_disable();
  skb = __skb_dequeue(&queue->input_pkt_queue);
  if (!skb) {
  __napi_complete(napi);
  local_irq_enable();
  break;
  }
  local_irq_enable();
  netif_receive_skb(skb);
} while (++work < quota && jiffies == start_time);
return work;
}
        首先，必须明白，process_backlog是在软中断上下文中被调用的，它有可能被外部中断打断。而queue->input_pkt_queue这样一个字段正处于和外部中断共享的临界区中，为避免竞争，这里需要将本地中断关闭。但是从另外一个角度，backlog的enqueue者似乎只有netif_rx，不会和外部中断发生竞争？如果是那样，这里的local_irq_disable是否可以省略？
        在关闭本地中断的情况下，从backlog的input_pkt_queue队列中取一个包，如果取不到，那么说明队列已经没有包需要处理，调用__napi_complete(napi)将backlog从softnet_data的poll_list中移除。这和napi_schedule是对应的。如果能够取到，那么自然会调用netif_receive_skb来收包。
        注意循环的退出条件，当process_backlog给定的quota用尽时，或者处理时间超过一个jiffies时，都会退出backlog处理。当前backlog napi的poll时间最多只有一个jiffies。
        另外，在当前CPU offline时被转移到其他cpu的softnet_data上。


二、NAPI方法
        NAPI的特点是进一次中断，可以处理多个包，所以设备要想使用NAPI，必须满足两个条件：
        （1）设备必须能够保留多个RX分组，例如RX DMA环形缓冲区。
        （2）设备必须能够禁用用于分组接受的IRQ。
        满足这两个条件，才能够做到在收包过程中不被外部中断打断，可以安心地将RX环形缓冲区中的分组转移到内存中。不过为了节约时间，大多数的驱动不会把驱动中的分组(buffer)拷贝到内存中，而只是将这些待处理的分组标记为CPU处理中。这样当外部设备接受分组的时候，检查到buffer被CPU使用，就不会再使用了。如果所有buffer都被CPU使用，就是环形缓冲区被占满了，驱动可以选择直接将新到来的分组丢弃。
        下面分析软中断NETIF_RX_IRQ的处理过程。

static void net_rx_action(struct softirq_action *h)
{
struct list_head *list = &__get_cpu_var(softnet_data).poll_list;
unsigned long time_limit = jiffies + 2;
int budget = netdev_budget;
void *have;
//1
local_irq_disable();
while (!list_empty(list)) {
  struct napi_struct *n;
  int work, weight;
//2
  if (unlikely(budget <= 0 || time_after(jiffies, time_limit)))
  goto softnet_break;
//3
  local_irq_enable();
  n = list_entry(list->next, struct napi_struct, poll_list);
  have = netpoll_poll_lock(n);
//4
  weight = n->weight;
  work = 0;
  if (test_bit(NAPI_STATE_SCHED, &n->state))
  work = n->poll(n, weight);
  WARN_ON_ONCE(work > weight);
  budget -= work;
//5
  local_irq_disable();
  if (unlikely(work == weight)) {
  if (unlikely(napi_disable_pending(n)))
   __napi_complete(n);
  else
   list_move_tail(&n->poll_list, list);
  }
  netpoll_poll_unlock(have);
}
out:
local_irq_enable();
return;

softnet_break:
__get_cpu_var(netdev_rx_stat).time_squeeze++;
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
goto out;
}
        （1）遍历softnet_data的轮询表，检查是否有napi实例需要处理。list_empty需要在关闭中断的情况下调用。
        （2）NET_RX_SOFTIRQ软中断每次能够处理的分组个数是有限制的，即netdev_budget。该值默认为300，但是可以在/proc/sys/net/core/netdev_budget中设置。另外，该软中断的处理时间不超过2个jiffies，否则就应该goto到softnet_break退出。由于是时间或者个数限制导致的退出，软中断poll_list上尚有分组没有处理完毕，因此需要重新触发NET_RX_SOFTIRQ。
        （3）list_entry可以在中断上下文中执行，而不会和软中断处理流程竞争，因此这里可以将中断打开。之后找到对应napi实例。
        （4）每一个napi实例能够处理的分组个数也是有限制的，这个限制即napi实例的weight。根据高速设备和低速设备的区分，这个值可以不同。例如高速设备的值为64，低速设备的值为16。之后调用napi实例的poll函数处理它所有分组。比如我们前面的process_backlog即时一个poll的示例。在process_backlog函数中，多个RX分组都保存在softnet_data->input_pkt_queue上。
        （5）一次poll就用尽了给它分配的所有weight，那么就要考虑将这个napi实例移到软中断轮询表的最后面，等待下次调用。给其他的napi实例一些机会。