关于NAPI的一些理解

在看e1000驱动的时候，顺便看了一下NAPI，了解了一下NAPI的原理。
注意:分析的内核源代码为2.6.18。

记得以前组成原理的课程上讲到过中断与轮询两种方式的IO方法，以为NAPI就是简单的轮询，直接一个while(1)循环，探测设备状态，读取数据。今天，跟踪代码，才发现实际上NAPI也是中断驱动的。是不是感觉有些怪异，中断驱动，那么跟传统的中断IO方法有什么区别呢？

传统中断IO
传统的中断IO方法，大家都比较熟悉，拿intel的千兆网卡举例来讲。网卡接收数据包，将数据包DMA到环形数据缓冲区后，触发中断，然后执行中断处理程序。中断处理程序，将数据包推向上层协议栈。
流程：
e1000_intr
e1000_clean_rx_irq
e1000_receive_skb
netif_rx

netif_rx:
1704int netif_rx(struct sk_buff *skb)
1705{
1706        struct softnet_data *queue;
1707        unsigned long flags;
1708
1709        /* if netpoll wants it, pretend we never saw it */
1710        if (netpoll_rx(skb))
1711                return NET_RX_DROP;
1712
1713        if (!skb->tstamp.tv64)
1714                net_timestamp(skb);
1715
1716        /*
1717         * The code is rearranged so that the path is the most
1718         * short when CPU is congested, but is still operating.
1719         */
1720        local_irq_save(flags);
1721        queue = &__get_cpu_var(softnet_data);
1722
1723        __get_cpu_var(netdev_rx_stat).total++;
1724        if (queue->input_pkt_queue.qlen <= netdev_max_backlog) {
1725                if (queue->input_pkt_queue.qlen) {
1726enqueue:
1727                        dev_hold(skb->dev);
1728                        __skb_queue_tail(&queue->input_pkt_queue, skb);
1729                        local_irq_restore(flags);
1730                        return NET_RX_SUCCESS;
1731                }
1732
1733                netif_rx_schedule(&queue->backlog_dev);
1734                goto enqueue;
1735        }
1736
1737        __get_cpu_var(netdev_rx_stat).dropped++;
1738        local_irq_restore(flags);
1739
1740        kfree_skb(skb);
1741        return NET_RX_DROP;
1742}

跟踪netif_rx_schedule函数:
netif_rx_schedule
__netif_rx_schedule
__raise_softirq_irqoff(NET_RX_SOFTIRQ);

这里触发了软件中断，NET_RX_SOFTIRQ在net_dev_init中初始化:
    open_softirq(NET_RX_SOFTIRQ, net_rx_action, NULL);
也就是说NET_RX_SOFTIRQ的软中断处理函数为net_rx_action。
1903static void net_rx_action(struct softirq_action *h)
1904{
1905        struct softnet_data *queue = &__get_cpu_var(softnet_data);
1906        unsigned long start_time = jiffies;
1907        int budget = netdev_budget;
1908        void *have;
1909
1910        local_irq_disable();
1911
1912        while (!list_empty(&queue->poll_list)) {
1913                struct net_device *dev;
1914
1915                if (budget <= 0 || jiffies - start_time > 1)
1916                        goto softnet_break;
1917
1918                local_irq_enable();
1919
1920                dev = list_entry(queue->poll_list.next,
1921                                 struct net_device, poll_list);
1922                have = netpoll_poll_lock(dev);
1923
1924                if (dev->quota <= 0 || dev->poll(dev, &budget)) {
1925                        netpoll_poll_unlock(have);
1926                        local_irq_disable();
1927                        list_move_tail(&dev->poll_list, &queue->poll_list);
1928                        if (dev->quota < 0)
1929                                dev->quota += dev->weight;
1930                        else
1931                                dev->quota = dev->weight;
1932                } else {
1933                        netpoll_poll_unlock(have);
1934                        dev_put(dev);
1935                        local_irq_disable();
1936                }
1937        }
1938out:
1939#ifdef CONFIG_NET_DMA
1940        /*
1941         * There may not be any more sk_buffs coming right now, so push
1942         * any pending DMA copies to hardware
1943         */
1944        if (net_dma_client) {
1945                struct dma_chan *chan;
1946                rcu_read_lock();
1947                list_for_each_entry_rcu(chan, &net_dma_client->channels, client_node)
1948                        dma_async_memcpy_issue_pending(chan);
1949                rcu_read_unlock();
1950        }
1951#endif
1952        local_irq_enable();
1953        return;
1954
1955softnet_break:
1956        __get_cpu_var(netdev_rx_stat).time_squeeze++;
1957        __raise_softirq_irqoff(NET_RX_SOFTIRQ);
1958        goto out;
1959}

从流程上来看，传统的IO方法是在中断处理函数中读取数据包。


NAPI
NAPI 是 Linux 上采用的一种提高网络处理效率的技术，它的核心概念就是不采用中断的方式读取数据，而代之以首先采用中断唤醒数据接收的服务程序，然后 POLL 的方法来轮询数据。虽然，读取数据不在中断中，并不意味着没有中断了。中断还是有的，并且还会触发中断处理函数。

NAPI使用的先决条件：
A. 要使用 DMA 的环形输入队列（也就是 ring_dma，这个在 2.4 驱动中关于 Ethernet 的部分有详细的介绍），或者是有足够的内存空间缓存驱动获得的包。
B. 在发送/接收数据包产生中断的时候有能力关断 NIC 中断的事件处理，并且在关断 NIC 以后，并不影响数据包接收到网络设备的环形缓冲区（以下简称 rx-ring）处理队列中。
C． 有防止 NIC 队列中排队的数据包冲突的能力。

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

跟踪NAPI的处理流程：
e1000_intr
__netif_rx_schedule(netdev);

这边就跟上面传统中断IO模式的后面处理一样了，触发软中断，执行dev->poll函数。

实际上NAPI，关键的一点就是系统在关中断的情况下，还能够接收数据包，当然这需要网卡的硬件支持。
注意：这里的关中断并不是CPU的关中断，而是屏蔽网卡的中断，一般是通过设置网卡中的中断掩码寄存器来实现的。中断掩码寄存器设置后，网卡依然可以通过DMA接收数据包，只是无法发送中断，来向CPU通知网卡接收事件而已。当然，在中断处理函数最后，需要开中断（设置中断屏蔽寄存器）。在新的Intel千兆网卡中，支持自动设置中断掩码（interrupt automask：我理解为发出中断后，自动设置中断屏蔽寄存器，屏蔽中断；在网卡的poll函数(e1000_clean)中，会在最后调用e1000_irq_enable，重新开中断）。

080813:
关于interrupt auto-mask可以查看我的另外一篇文章:
intel千兆网卡的interrupt auto-mask分析
http://blog.chinaunix.net/u/12592/showart_1129657.html

相比之下，对于传统的中断IO方式，不需要自己对设备关中断和开中断（不是CPU的关中断和开中断指令。当外部发生中断时，硬件会自动关闭中断；当从中断处理程序返回时，在POP出FLAG寄存器时，就自动开中断了）。NAPI方式需要一个关中断的环境（网卡接受数据，但是不产生中断），因此需要在设置自己的中断屏蔽寄存器来进行屏蔽中断。
在e1000_probe函数中，设置ICR.INT_ASSERTED，来激活interrupt automask:
    /* Hardware features, flags and workarounds */
    if (adapter->hw.mac.type >= e1000_82571) {
        adapter->flags.int_assert_auto_mask = 1;
#ifdef CONFIG_PCI_MSI
        adapter->flags.has_msi = 1;
#endif
        adapter->flags.has_manc2h = 1;
    }

Interrupt asserted (ICR.INT_ASSERTED = 1b) - ICR content is cleared and Auto
Mask is active (the IAM register is written to the IMC register).

当关断发送/接收事件中断的时候，NAPI 将在 POLL 中被调用处理，由于 POLL 方法的时候，NIC 中断已经不能通知包到达，那么这个时候在如果在完成轮询，并且中断打开以后，会马上有一个 NIC 中断产生，从而触发一次 POLL 事件，这种在中断关断时刻到达的包我们称为"rotting"；这样就会在 POLL 机制和 NIC 中断之间产生一个竞争，解决的方法就是利用网卡的接收状态位，继续接收环形队列缓冲 rx-ring 中的数据，直到没有数据接收以后，才使能中断。


兼容NAPI
softnet_data有个backlog_dev成员，其原因就是为了兼容在NAPI。
最新的内核的网卡驱动的编程接口都已经变为了NAPI的方式了， NAPI接口是以poll函数为核心的。但旧的netif_rx接口也要兼容，最糟糕的是旧的驱动模式中没有实现poll函数的硬性要求，怎么办呢？于是内核的编程者就想了个折衷的权宜之计，即在 softnet_data增加一个backlog_dev，意思就是后备的没有办法的方法，虽然这样做很ugly ：）。
思想明白了，编程就很简单了。
现在看代码：
在实现了新的NAPI的网卡驱动程序中，中断函数要先调用netif_rx_schedule(),这
个函数做的事就讲本dev 挂到本CPU的 softnet_data成员poll_list中：
1118void __netif_rx_schedule(struct net_device *dev)
1119{
1120        unsigned long flags;
1121
1122        local_irq_save(flags);
1123        dev_hold(dev);
1124        list_add_tail(&dev->poll_list, &__get_cpu_var(softnet_data).poll_list);
1125        if (dev->quota < 0)
1126                dev->quota += dev->weight;
1127        else
1128                dev->quota = dev->weight;
1129        __raise_softirq_irqoff(NET_RX_SOFTIRQ);
1130        local_irq_restore(flags);
1131}
1132EXPORT_SYMBOL(__netif_rx_schedule);

注意上面红色标注的两个netif_rx_schedule的地方，可以发现两个的参数是不同的。
在软中断的net_rx_action中的softnet_data的成员poll_list中存放的是就是本dev了，而不是backlog_dev；在兼容就的接口netif_rx时，因为网卡的net_device没有poll函数，netif_rx只得把包挂到softnet_data的队列中， 并且借用backlog_dev的poll函数来处理，以达到从宏观上看所有的驱动（包括旧的netif_rx接口的旧驱动）都实现了NAPI。


参考：
http://www.ibm.com/developerworks/cn/linux/l-napi/

http://hi.baidu.com/linux_kernel/blog/item/b2919a506d11b46284352418.html

http://hi.baidu.com/zengzhaonong/blog/item/c48864f4ec5401daf3d3856e.html