Linux的NAPI方法收包_netif_rx返回值

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实例一些机会。