五、队列层1、软中断与下半部
当用中断处理的时候,为了减少中断处理的工作量,比如,一般中断处理时,需要屏蔽其它中断,如果中断处理时间过长,那么其它中断
有可能得不到及时处理,也以,有一种机制,就是把“不必马上处理”的工作,推迟一点,让它在中断处理后的某一个时刻得到处理。这就
是下半部。
下半部只是一个机制,它在Linux中,有多种实现方式,其中一种对时间要求最严格的实现方式,叫“软中断”,可以使用:
open_softirq()
来向内核注册一个软中断,
然后,在合适的时候,调用
raise_softirq_irqoff()
触发它。
如果采用中断方式接收数据(这一节就是在说中断方式接收,后面,就不用这种假设了),同样也需要软中断,可以调用
open_softirq(NET_RX_SOFTIRQ, net_rx_action, NULL);
向内核注册一个名为NET_RX_SOFTIR的软中断,net_rx_action是软中断的处理函数。
然后,在驱动中断处理完后的某一个时刻,调用
raise_softirq_irqoff(NET_RX_SOFTIRQ);
触发它,这样net_rx_action将得到执行。
2、队列层
什么是队列层?通常,在网卡收发数据的时候,需要维护一个缓冲区队列,来缓存可能存在的突发数据,类似于前面的DMA环形缓冲区。
队列层中,包含了一个叫做struct softnet_data:
struct softnet_data
{
/*throttle 用于拥塞控制,当拥塞发生时,throttle将被设置,后续进入的数据包将被丢弃*/
int throttle;
/*netif_rx函数返回的拥塞级别*/
int cng_level;
int avg_blog;
/*softnet_data 结构包含一个指向接收和传输队列的指针,input_pkt_queue成员指向准备传送
给网络层的sk_buffs包链表的首部的指针,这个队列中的包是由netif_rx函数递交的*/
struct sk_buff_head input_pkt_queue;
struct list_head poll_list;
struct net_device *output_queue;
struct sk_buff *completion_queue;
struct net_device backlog_dev; /* Sorry. 8) */
};
内核使用了一个同名的变量softnet_data,它是一个Per-CPU变量,每个CPU都有一个。
net/core/dev.cDECLARE_PER_CPU(struct softnet_data,softnet_data);
/*
* 网络模块的核心处理模块.
*/
static int __init net_dev_init(void)
{
int i, rc = -ENOMEM;
BUG_ON(!dev_boot_phase);
net_random_init();
if (dev_proc_init()) /*初始化proc文件系统*/
goto out;
if (netdev_sysfs_init()) /*初始化sysfs文件系统*/
goto out;
/*ptype_all和ptype_base是重点,后面会详细分析,它们都是
struct list_head类型变量,这里初始化链表成员*/
INIT_LIST_HEAD(&ptype_all);
for (i = 0; i < 16; i++)
INIT_LIST_HEAD(&ptype_base[i]);
for (i = 0; i < ARRAY_SIZE(dev_name_head); i++)
INIT_HLIST_HEAD(&dev_name_head[i]);
for (i = 0; i < ARRAY_SIZE(dev_index_head); i++)
INIT_HLIST_HEAD(&dev_index_head[i]);
/*
* 初始化包接收队列,这里我们的重点了.
*/
/*遍历每一个CPU,取得它的softnet_data,我们说过,它是一个struct softnet_data的Per-CPU变量*/
for (i = 0; i < NR_CPUS; i++) {
struct softnet_data *queue;
/*取得第i个CPU的softnet_data,因为队列是包含在它里边的,所以,我会直接说,“取得队列”*/
queue = &per_cpu(softnet_data, i);
/*初始化队列头*/
skb_queue_head_init(&queue->input_pkt_queue);
queue->throttle = 0;
queue->cng_level = 0;
queue->avg_blog = 10; /* arbitrary non-zero */
queue->completion_queue = NULL;
INIT_LIST_HEAD(&queue->poll_list);
set_bit(__LINK_STATE_START, &queue->backlog_dev.state);
queue->backlog_dev.weight = weight_p;
/*这里,队列中backlog_dev设备,它是一个伪网络设备,不对应任何物理设备,它的poll函数,指向了
process_backlog,后面我们会详细分析*/
queue->backlog_dev.poll = process_backlog;
atomic_set(&queue->backlog_dev.refcnt, 1);
}
#ifdef OFFLINE_SAMPLE
samp_timer.expires = jiffies + (10 * HZ);
add_timer(&samp_timer);
#endif
dev_boot_phase = 0;
/*注册收/发软中断*/
open_softirq(NET_TX_SOFTIRQ, net_tx_action, NULL);
open_softirq(NET_RX_SOFTIRQ, net_rx_action, NULL);
hotcpu_notifier(dev_cpu_callback, 0);
dst_init();
dev_mcast_init();
rc = 0;
out:
return rc;
}
这样,初始化完成后,在驱动程序中,在中断处理函数中,会调用netif_rx将数据交上来,这与采用轮询技术,有本质的不同:
int netif_rx(struct sk_buff *skb)
{
int this_cpu;
struct softnet_data *queue;
unsigned long flags;
/* if netpoll wants it, pretend we never saw it */
if (netpoll_rx(skb))
return NET_RX_DROP;
/*接收时间戳未设置,设置之*/
if (!skb->stamp.tv_sec)
net_timestamp(&skb->stamp);
/*
* 这里准备将数据包放入接收队列,需要禁止本地中断,在入队操作完成后,再打开中断.
*/
local_irq_save(flags);
/*获取当前CPU对应的softnet_data变量*/
this_cpu = smp_processor_id();
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) {
if (queue->throttle) /*拥塞发生了,丢弃数据包*/
goto drop;
/*数据包入队操作*/
enqueue:
dev_hold(skb->dev); /*累加设备引入计数器*/
__skb_queue_tail(&queue->input_pkt_queue, skb); /*将数据包加入接收队列*/
#ifndef OFFLINE_SAMPLE
get_sample_stats(this_cpu);
#endif
local_irq_restore(flags);
return queue->cng_level;
}
/*
* 驱动程序不断地调用net_rx函数,实现接收数据包的入队操作,当qlen == 0时, 则进入这段代码,这里,如果已经被设置拥塞标志的话,则清除它,因为这里将要调用软中断,开始将数据包交给 上层了,即上层协议的接收函数将执行出队操作,拥塞自然而然也就不存在了。 */
if (queue->throttle)
queue->throttle = 0;
/*
* netif_rx_schedule函数完成两件重要的工作:
* 1、将bakclog_dev设备加入“处理数据包的设备”的链表当中;
* 2、触发软中断函数,进行数据包接收处理;
*/
netif_rx_schedule(&queue->backlog_dev);
goto enqueue;
}
/*前面判断了队列是否已满,如果已满而标志未设置,设置之,并累加拥塞计数器*/
if (!queue->throttle) {
queue->throttle = 1;
__get_cpu_var(netdev_rx_stat).throttled++;
}
/*拥塞发生,累加丢包计数器,释放数据包*/
drop:
__get_cpu_var(netdev_rx_stat).dropped++;
local_irq_restore(flags);
kfree_skb(skb);
return NET_RX_DROP;
}
从这段代码的分析中,我们可以看到,当第一个数据包被接收后,因为qlen==0,所以首先会调用netif_rx_schedule触发软中断,然后利用goto跳转至入队。因为软中断被触发后,将执行出队操作,把数据交往上层处理。而当这个时候,又有数据包进入,即网卡中断产生,因为它的优先级高过软中断,这样,出队操作即被中断,网卡中断程序再将被调用,netif_rx函数又再次被执行,如果队列未满,就入队返回。中断完成后,软中断的执行过程被恢复而继续执行出队——如此生产者/消费者循环不止,生生不息……
netif_rx调用netif_rx_schedule进一步处理数据包,我们注意到:
1、前面讨论过,采用轮询技术时,同样地,也是调用netif_rx_schedule,把设备自己传递了过去;
2、这里,采用中断方式,传递的是队列中的一个“伪设备”,并且,这个伪设备的poll函数指针,指向了一个叫做process_backlog的函数;
netif_rx_schedule函数完成两件重要的工作:
1、将bakclog_dev设备加入“处理数据包的设备”的链表当中;
2、触发软中断函数,进行数据包接收处理;
这样,我们可以猜想,在软中断函数中,不论是伪设备bakclog_dev,还是真实的设备(如前面讨论过的e100),都会被软中断函数以:
dev->poll()
的形式调用,对于e100来说,poll函数的接收过程已经分析了,而对于其它所有没有采用轮询技术的网络设备来说,它们将统统调用
process_backlog函数(我觉得把它改名为pseudo-poll是否更合适一些^o^)。
OK,我想分析到这里,关于中断处理与轮询技术的差异,已经基本分析开了……
继续来看,netif_rx_schedule进一步调用__netif_rx_schedule:
/* Try to reschedule poll. Called by irq handler. */
static inline void netif_rx_schedule(struct net_device *dev)
{
if (netif_rx_schedule_prep(dev))
__netif_rx_schedule(dev);
}
/* Add interface to tail of rx poll list. This assumes that _prep has
* already been called and returned 1.
*/
static inline void __netif_rx_schedule(struct net_device *dev)
{
unsigned long flags;
local_irq_save(flags);
dev_hold(dev);
/*伪设备也好,真实的设备也罢,都被加入了队列层的设备列表*/
list_add_tail(&dev->poll_list, &__get_cpu_var(softnet_data).poll_list);
if (dev->quota < 0)
dev->quota += dev->weight;
else
dev->quota = dev->weight;
/*触发软中断*/
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
local_irq_restore(flags);
}
软中断被触发,注册的net_rx_action函数将被调用:
/*接收的软中断处理函数*/
static void net_rx_action(struct softirq_action *h)
{
struct softnet_data *queue = &__get_cpu_var(softnet_data);
unsigned long start_time = jiffies;
int budget = netdev_max_backlog;
local_irq_disable();
/*
* 遍历队列的设备链表,如前所述,__netif_rx_schedule已经执行了
* list_add_tail(&dev->poll_list, &__get_cpu_var(softnet_data).poll_list);
* 设备bakclog_dev已经被添加进来了
*/
while (!list_empty(&queue->poll_list)) {
struct net_device *dev;
if (budget <= 0 || jiffies - start_time > 1)
goto softnet_break;
local_irq_enable();
/*取得链表中的设备*/
dev = list_entry(queue->poll_list.next,
struct net_device, poll_list);
netpoll_poll_lock(dev);
/*调用设备的poll函数,处理接收数据包,这样,采用轮询技术的网卡,它的真实的poll函数将被调用,
这就回到我们上一节讨论的e100_poll函数去了,而对于采用传统中断处理的设备,它们调用的,都将是
bakclog_dev的process_backlog函数*/
if (dev->quota <= 0 || dev->poll(dev, &budget)) {
netpoll_poll_unlock(dev);
/*处理完成后,把设备从设备链表中删除,又重置于末尾*/
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 {
netpoll_poll_unlock(dev);
dev_put(dev);
local_irq_disable();
}
}
out:
local_irq_enable();
return;
softnet_break:
__get_cpu_var(netdev_rx_stat).time_squeeze++;
__raise_softirq_irqoff(NET_RX_SOFTIRQ);
goto out;
}
对于dev->poll(dev, &budget)的调用,一个真实的poll函数的例子,我们已经分析过了,现在来看process_backlog,
static int process_backlog(struct net_device *backlog_dev, int *budget)
{
int work = 0;
int quota = min(backlog_dev->quota, *budget);
struct softnet_data *queue = &__get_cpu_var(softnet_data);
unsigned long start_time = jiffies;
backlog_dev->weight = weight_p;
/*在这个循环中,执行出队操作,把数据从队列中取出来,交给netif_receive_skb,直至队列为空*/
for (;;) {
struct sk_buff *skb;
struct net_device *dev;
local_irq_disable();
skb = __skb_dequeue(&queue->input_pkt_queue);
if (!skb)
goto job_done;
local_irq_enable();
dev = skb->dev;
netif_receive_skb(skb);
dev_put(dev);
work++;
if (work >= quota || jiffies - start_time > 1)
break;
}
backlog_dev->quota -= work;
*budget -= work;
return -1;
/*当队列中的数据包被全部处理后,将执行到这里*/
job_done:
backlog_dev->quota -= work;
*budget -= work;
list_del(&backlog_dev->poll_list);
smp_mb__before_clear_bit();
netif_poll_enable(backlog_dev);
if (queue->throttle)
queue->throttle = 0;
local_irq_enable();
return 0;
}
这个函数重要的工作,就是出队,然后调用netif_receive_skb()将数据包交给上层,这与上一节讨论的poll是一样的。这也是为什么,
在网卡驱动的编写中,采用中断技术,要调用netif_rx,而采用轮询技术,要调用netif_receive_skb啦!
到了这里,就处理完数据包与设备相关的部分了,数据包将进入上层协议栈……