handle_dev_cpu_collision
来自 ./net/sched/sch_generic.c 的代码 handle_dev_cpu_collision 处理两种情况：
- 传输锁由当前 CPU 持有。
- 传输锁由其他 CPU 持有。
在第一种情况下，这被作为配置问题处理，因此打印警告。 在第二种情况下，增加统计计数器cpu_collision，并且数据经 dev_requeue_skb 发送，以便稍后重新排队传输。 回想一下，我们在 dequeue_skb 中看到的专门处理重新排队的 skb 代码。
handle_dev_cpu_collision 的代码很短，值得快速阅读：
static inline int handle_dev_cpu_collision(struct sk_buff *skb, struct netdev_queue *dev_queue, struct Qdisc *q) { int ret; if (unlikely(dev_queue->xmit_lock_owner == smp_processor_id())) { /* * Same CPU holding the lock. It may be a transient * configuration error, when hard_start_xmit() recurses. We * detect it by checking xmit owner and drop the packet when * deadloop is detected. Return OK to try the next skb. */ kfree_skb(skb); net_warn_ratelimited("Dead loop on netdevice %s, fix it urgently!\n", dev_queue->dev->name); ret = qdisc_qlen(q); } else { /* * Another cpu is holding lock, requeue & delay xmits for * some time. */ __this_cpu_inc(softnet_data.cpu_collision); ret = dev_requeue_skb(skb, q); } return ret; }
让我们来看看 dev_requeue_skb 做了什么，因为我们将看到这个函数是从 sch_direct_xmit 调用的。
dev_requeue_skb
值得庆幸的是，dev_requeue_skb 的源代码很短，而且直截了当，来自 ./net/sched/sch_generic.c：
/* Modifications to data participating in scheduling must be protected with * qdisc_lock(qdisc) spinlock. * * The idea is the following: * - enqueue, dequeue are serialized via qdisc root lock * - ingress filtering is also serialized via qdisc root lock * - updates to tree and tree walking are only done under the rtnl mutex. */ static inline int dev_requeue_skb(struct sk_buff *skb, struct Qdisc *q) { skb_dst_force(skb); q->gso_skb = skb; q->qstats.requeues++; q->q.qlen++; /* it's still part of the queue */ __netif_schedule(q); return 0; }
这个函数做了几件事：
- 它强制增加 skb 引用计数。
- 它关联 skb 到 qdisc 的
gso_skb字段。 回想一下，我们之前看到，在从 qdisc 的队列中取出数据之前，会在dequeue_skb中检查此字段。 - 增加统计计数器。
- 增加队列的大小。
- 调用
__netif_schedule。
简单明了。 让我们回顾一下我们是如何到达这里的，然后探讨 __netif_schedule。
提醒， __qdisc_run 中的 while 循环
回想一下，我们是检查函数 __qdisc_run 得出的这一点，该函数包含以下代码：
void __qdisc_run(struct Qdisc *q) { int quota = weight_p; while (qdisc_restart(q)) { /* * Ordered by possible occurrence: Postpone processing if * 1. we've exceeded packet quota * 2. another process needs the CPU; */ if (--quota <= 0 || need_resched()) { __netif_schedule(q); break; } } qdisc_run_end(q); }
这段代码的工作原理是在一个循环中反复调用 qdisc_restart，在内部，它会使 skb 出队，并试图调用 sch_direct_xmit 来传输 skb，而 sch_direct_xmit 会调用 dev_hard_start_xmit 来执行实际的传输。 任何不能传输的内容都将在 NET_TX 软中断中重新排队以进行传输。
传输过程中的下一步是检查 dev_hard_start_xmit，以了解如何调用驱动程序来发送数据。 在此之前，我们应该研究 __netif_schedule 以完全理解 __qdisc_run 和 dev_requeue_skb 是如何工作的。
__netif_schedule
让我们从 ./net/core/dev.c 跳到 __netif_schedule：
void __netif_schedule(struct Qdisc *q) { if (!test_and_set_bit(__QDISC_STATE_SCHED, &q->state)) __netif_reschedule(q); } EXPORT_SYMBOL(__netif_schedule);
此代码检查并设置 qdisc 状态的 __QDISC_STATE_SCHED 位。 如果该位被翻转（意味着它之前没有处于 __QDISC_STATE_SCHED 状态），代码将调用 __netif_reschedule，这并不长，但有非常有趣的附带作用。 我们来看一下：
static inline void __netif_reschedule(struct Qdisc *q) { struct softnet_data *sd; unsigned long flags; local_irq_save(flags); sd = &__get_cpu_var(softnet_data); q->next_sched = NULL; *sd->output_queue_tailp = q; sd->output_queue_tailp = &q->next_sched; raise_softirq_irqoff(NET_TX_SOFTIRQ); local_irq_restore(flags); }
此函数执行以下操作：
- 保存当前的本地 IRQ 状态，并调用
local_irq_save禁用 IRQ。 - 获取当前 CPU
softnet_data结构。 - 添加 qdisc 到
softnet_data的输出队列。 - 触发
NET_TX_SOFTIRQ软中断。 - 恢复 IRQ 状态并重新启用中断。
你可以阅读我们之前关于网络栈接收端的文章，来了解更多关于 softnet_data 数据结构初始化的信息。
上面函数中的重要代码是：raise_softirq_irqoff 触发 NET_TX_SOFTIRQ 软中断。softirq 及其注册也在我们的前一篇文章中介绍过。 简单地说，您可以认为软中断是内核线程，它们以非常高的优先级执行，并代表内核处理数据。 它们处理传入的网络数据，也处理传出的数据。
正如你在上一篇文章中看到的，NET_TX_SOFTIRQ 软中断注册了函数 net_tx_action。这意味着有一个内核线程在执行 net_tx_action。 该线程偶尔会暂停，raise_softirq_irqoff 会恢复它。让我们来看看 net_tx_action 是做什么的，这样我们就可以理解内核是如何处理传输请求的。
net_tx_action
net_tx_action 函数位于 ./net/core/dev.c 文件中，它在运行时处理两个主要内容：
- 执行 CPU 的
softnet_data结构的完成队列。 - 执行 CPU 的
softnet_data结构的输出队列。
实际上，该函数的代码是两个大的 if 块。 让我们一次查看一个，同时记住这段代码是作为一个独立的内核线程在软中断上下文中执行的。 net_tx_action 的目的是在整个网络对战的传输侧执行不能在热点路径中执行的代码；工作被延迟，稍后由执行 net_tx_action 的线程进行处理。
net_tx_action 完成队列
softnet_data 的完成队列只是一个等待释放的 skb 队列。 函数 dev_kfree_skb_irq 添加 skb 到队列中以便稍后释放。 设备驱动程序通常使用此选项来延迟释放已使用的 skb。 驱动程序希望延迟释放 skb 而不是简单地释放 skb，原因是释放内存可能需要时间，在某些实例（如 hardirq 处理程序）中，代码需要尽可能快地执行并返回。
看一下 net_tx_action 代码，它处理在完成队列上释放 skb：
if (sd->completion_queue) { struct sk_buff *clist; local_irq_disable(); clist = sd->completion_queue; sd->completion_queue = NULL; local_irq_enable(); while (clist) { struct sk_buff *skb = clist; clist = clist->next; WARN_ON(atomic_read(&skb->users)); trace_kfree_skb(skb, net_tx_action); __kfree_skb(skb); } }
如果完成队列有条目，while 循环将遍历 skb 的链表，并对每个 skb 调用 __kfree_skb 以释放它们的内存。 请记住，这段代码是在一个单独的“线程”中运行的，该线程名为 softirq – 它并不代表任何特定的用户程序运行。
net_tx_action 输出队列
输出队列的用途完全不同。 如前所述，调用 __netif_reschedule 添加数据到输出队列，该调用通常从 __netif_schedule 调用的。 到目前为止，在我们在两个实例中看到过调用了 __netif_schedule 函数：
dev_requeue_skb：正如我们所看到的，如果驱动程序报告错误码NETDEV_TX_BUSY或 CPU 冲突，则可以调用此函数。__qdisc_run：我们之前也看到过这个函数。 一旦超过配额或需要重新调度进程，它还会调用__netif_schedule。
在这两种情况下，都将调用 __netif_schedule 函数，该函数添加 qdisc 到 softnet_data 的输出队列中进行处理。 我将输出队列处理代码分成了三个块。 我们先来看看第一个：
if (sd->output_queue) { struct Qdisc *head; local_irq_disable(); head = sd->output_queue; sd->output_queue = NULL; sd->output_queue_tailp = &sd->output_queue; local_irq_enable();
这个块只是确保输出队列上有 qdisc，如果有，它设置 head 为第一个条目，并移动队列的尾指针。
接下来，遍历 qdsics 列表的 while 循环开始：
while (head) { struct Qdisc *q = head; spinlock_t *root_lock; head = head->next_sched; root_lock = qdisc_lock(q); if (spin_trylock(root_lock)) { smp_mb__before_clear_bit(); clear_bit(__QDISC_STATE_SCHED, &q->state); qdisc_run(q); spin_unlock(root_lock);
上面的代码段向前移动头指针，并获得对 qdisc 锁的引用。spin_trylock 检查是否可以获得锁；注意，该调用是专门使用的，因为它不阻塞。 如果锁已经被持有，spin_trylock 将立即返回，而不是等待获得锁。
如果 spin_trylock 成功获得锁，则返回一个非零值。 在这种情况下，qdisc 的状态字段的__QDISC_STATE_SCHED 位翻转，qdisc_run 被调用，从而翻转 __QDISC___STATE_RUNNING位，并开始执行 __qdisc_run。
这很重要。这里发生的情况是，我们之前检查过的代表用户进行系统调用的处理循环，现在再次运行，但在 softirq 上下文中，因为此 qdisc 的 skb 传输无法传输。 这种区别很重要，因为它会影响您如何监控发送大量数据的应用程序的 CPU 使用情况。 让我换个方式说：
- 程序的系统时间包括调用驱动程序以尝试发送数据所花费的时间，无论发送是否完成或驱动程序是否返回错误。
- 如果在驱动程序层发送不成功（例如，因为设备忙于发送其他内容），则添加 qdisc 到输出队列并稍后由 softirq 线程处理。 在这种情况下，将花费 softirq（si）时间来尝试传输您的数据。
因此，发送数据所花费的总时间是与发送相关的系统调用的系统时间和 NET_TX 软中断的软中断时间的组合。
无论如何，上面的代码释放 qdisc 锁来完成。 如果上面获取锁的 spin_trylock 调用失败，则执行以下代码：
} else { if (!test_bit(__QDISC_STATE_DEACTIVATED, &q->state)) { __netif_reschedule(q); } else { smp_mb__before_clear_bit(); clear_bit(__QDISC_STATE_SCHED, &q->state); } } } }
这段代码只在无法获得 qdisc 锁时执行，它处理两种情况。 两者之一：
- 未停用 qdisc，但无法获取执行
qdisc_run的锁。 所以，调用__netif_reschedule。 在这里调用__netif_reschedule会将 qdisc 放回该函数当前出列的队列中。这允许在以后可能已经放弃锁时再次检查 qdisc。 - qdisc 被标记为停用，确保
__QDISC_STATE_SCHED状态标志也被清除。
