1. 如何在linux下开启napi
天重点对linux网络数据包的处理做下分析，但是并不关系到上层协议，仅仅到链路层。
之前转载过一篇文章，对NAPI做了比较详尽的分析，本文结合Linux内核源代码，对当前网络数据包的处理进行梳理。根据NAPI的处理特性，对设备提出一定的要求
1、设备需要有足够的缓冲区，保存多个数据分组
2、可以禁用当前设备中断，然而不影响其他的操作。
当前大部分的设备都支持NAPI，但是为了对之前的保持兼容，内核还是对之前中断方式提供了兼容。我们先看下NAPI具体的处理方式。我们都知道中断分为中断上半部和下半部，上半部完成的任务很是简单，仅仅负责把数据保存下来；而下半部负责具体的处理。为了处理下半部，每个CPU有维护一个softnet_data结构。我们不对此结构做详细介绍，仅仅描述和NAPI相关的部分。结构中有一个poll_list字段，连接所有的轮询设备。还 维护了两个队列input_pkt_queue和process_queue。这两个用户传统不支持NAPI方式的处理。前者由中断上半部的处理函数吧数据包入队，在具体的处理时，使用后者做中转，相当于前者负责接收，后者负责处理。最后是一个napi_struct的backlog，代表一个虚拟设备供轮询使用。在支持NAPI的设备下，每个设备具备一个缓冲队列，存放到来数据。每个设备对应一个napi_struct结构，该结构代表该设备存放在poll_list中被轮询。而设备还需要提供一个poll函数，在设备被轮询到后，会调用poll函数对数据进行处理。基本逻辑就是这样，下面看下具体流程。
中断上半部：
非NAPI：
非NAPI对应的上半部函数为netif_rx，位于Dev.,c中
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;
}
中间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;
}
/* Schele 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_schele(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_schele函数，该函数把设备对应的napi_struct结构插入到softnet_data的poll_list链表尾部，然后唤醒软中断，这样在下次软中断得到处理时，中断下半部就会得到处理。不妨看下源码
static inline void ____napi_schele(struct softnet_data *sd,
struct napi_struct *napi)
{
list_add_tail(&napi->poll_list, &sd->poll_list);
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
}
NAPI方式
NAPI的方式相对于非NAPI要简单许多，看下e100网卡的中断处理函数e100_intr，核心部分
if (likely(napi_schele_prep(&nic->napi))) {
e100_disable_irq(nic);//屏蔽当前中断
__napi_schele(&nic->napi);//把设备加入到轮训队列
}
if条件检查当前设备是否 可被调度，主要检查两个方面：1、是否已经在调度 2、是否禁止了napi pending.如果符合条件，就关闭当前设备的中断，调用__napi_schele函数把设备假如到轮训列表，从而开启轮询模式。
分析：结合上面两种方式，还是可以发现两种方式的异同。其中softnet_data作为主导结构，在NAPI的处理方式下，主要维护轮询链表。NAPI设备均对应一个napi_struct结构，添加到链表中；非NAPI没有对应的napi_struct结构，为了使用NAPI的处理流程，使用了softnet_data结构中的back_log作为一个虚拟设备添加到轮询链表。同时由于非NAPI设备没有各自的接收队列，所以利用了softnet_data结构的input_pkt_queue作为全局的接收队列。这样就处理而言，可以和NAPI的设备进行兼容。但是还有一个重要区别，在NAPI的方式下，首次数据包的接收使用中断的方式，而后续的数据包就会使用轮询处理了；而非NAPI每次都是通过中断通知。
下半部：
下半部的处理函数，之前提到，网络数据包的接发对应两个不同的软中断，接收软中断NET_RX_SOFTIRQ的处理函数对应net_rx_action
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();
/*遍历轮询表*/
while (!list_empty(&sd->poll_list)) {
struct napi_struct *n;
int work, weight;
/* If softirq window is exhuasted then punt.
* Allow this to run for 2 jiffies since which will allow
* an average latency of 1.5/HZ.
*/
/*如果开支用完了或者时间用完了*/
if (unlikely(budget <= 0 || time_after_eq(jiffies, time_limit)))
goto softnet_break;
local_irq_enable();
/* Even though interrupts have been re-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);
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 scheled.
*/
work = 0;
/*如果被设备已经被调度，则调用其处理函数poll函数*/
if (test_bit(NAPI_STATE_SCHED, &n->state)) {
work = n->poll(n, weight);//后面weight指定了一个额度
trace_napi_poll(n);
}
WARN_ON_ONCE(work > weight);
/*总额度递减*/
budget -= work;
local_irq_disable();
/* 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.
*/
/*如果work=weight的话。任务就完成了，把设备从轮询链表删除*/
if (unlikely(work == weight)) {
if (unlikely(napi_disable_pending(n))) {
local_irq_enable();
napi_complete(n);
local_irq_disable();
} else {
if (n->gro_list) {
/* flush too old packets
* If HZ < 1000, flush all packets.
*/
local_irq_enable();
napi_gro_flush(n, HZ >= 1000);
local_irq_disable();
}
/*每次处理完就把设备移动到列表尾部*/
list_move_tail(&n->poll_list, &sd->poll_list);
}
}
netpoll_poll_unlock(have);
}
out:
net_rps_action_and_irq_enable(sd);
#ifdef CONFIG_NET_DMA
/*
* There may not be any more sk_buffs coming right now, so push
* any pending DMA copies to hardware
*/
dma_issue_pending_all();
#endif
return;
softnet_break:
sd->time_squeeze++;
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
goto out;
}
这里有处理方式比较直观，直接遍历poll_list链表，处理之前设置了两个限制：budget和time_limit。前者限制本次处理数据包的总量，后者限制本次处理总时间。只有二者均有剩余的情况下，才会继续处理。处理期间同样是开中断的，每次总是从链表表头取设备进行处理，如果设备被调度，其实就是检查NAPI_STATE_SCHED位，则调用 napi_struct的poll函数，处理结束如果没有处理完，则把设备移动到链表尾部，否则从链表删除。NAPI设备对应的poll函数会同样会调用__netif_receive_skb函数上传协议栈，这里就不做分析了，感兴趣可以参考e100的poll函数e100_poll。
而非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。如果剩余额度足够处理完这些数据包，那么就把虚拟设备移除轮询队列。这里有些疑惑就是最后为何要增加额度，剩下的额度已经足够处理这些数据了呀？根据此流程不难发现，其实执行的是在两个队列之间移动数据包，然后再做处理。
2. 数组方法里push，pop，shift，unshift，join，splice分别是什么作用
数组操作函数有：push，pop，join，shift，unshift，slice，splice，concat
（1）push 和 pop
这两个函数都是对数组从尾部进行压入或弹出操作。push(arg1,arg2,...)可以每次压入一个或多个元素，并返回更新后的数组长度。注意如果参数也是数组的话，则是将全部数组当做一个元素压入到原本的数组里面去。pop() 函数则每次只会弹出结尾的元素，并返回弹出的元素，若是是对空组数调用 pop() 则返回undefined。
示例：
var oldArr=[1,2,3];
alert(oldArr.push(4,[5,6]))//这里只会将[5,6]当做一个元素来策画，返回更新后的数组长度5
此时 oldArr = [1,2,3,4,[5,6]]
oldArr.pop()//这里弹出最后一个元素[5,6]，而不是6
此时 oldArr = [1,2,3,4]
oldArr.pop()-->4
oldArr.pop()-->3
oldArr.pop()-->2
oldArr.pop()-->1
alert(oldArr.pop())-->undefined(空数组弹出)
（2）unshift 和 shift
unshift() 方法可向数组的开头添加一个或更多元素，并返回新的长度。unshift() 方法将把它的参数插入 arrayObject 的头部，并将已经存在的元素顺次地移到较高的下标处，以便留出空间。该方法的第一个参数将成为数组的新元素 0，如果还有第二个参数，它将成为新的元素 1，以此类推。
请注意，unshift() 方法不创建新的创建，而是直接修改原有的数组。另外在 Internet Explorer 浏览器中 unshift() 无法执行！