调优：Transmit Packet Steering(XPS)
要使 XPS 工作，必须在内核配置中启用它（在 Ubuntu 的内核 3.13.0 上是启用的），并且需要一个位掩码来描述哪些 CPU 应该处理给定接口和传输队列的数据包。
这些位掩码类似于 RPS 位掩码，您可以在内核文档中找到关于这些位掩码的一些 文档。
简而言之，要修改的位掩码位于：
/sys/class/net/DEVICE_NAME/queues/QUEUE/xps_cpus
因此，对于 eth0 和传输队列 0，您需要修改文件：/sys/class/net/eth0/queues/tx-0/xps_cpus，其中十六进制数指示哪些 CPU 应处理来自 eth0 的传输队列 0 的传输完成。 正如文档所指出的，XPS 在某些配置中可能是不必要的。
排队规则！
要了解网络数据的路径，我们需要稍微了解一下 qdisc 代码。本文不打算涵盖每个不同传输队列选项的具体细节。 如果你对此感兴趣，请查看这本优秀的指南。
在这篇博客文章中，我们将继续代码路径，研究通用包调度器代码是如何工作的。 特别是，我们将探索 qdisc_run_begin、qdisc_run_end、__qdisc_run 和 sch_direct_xmit 如何移动网络数据到更靠近传输驱动程序的位置。
让我们先看看 qdisc_run_begin 是如何工作的，并从那里开始。
qdisc_run_begin 和 qdisc_run_end
qdisc_run_begin 函数可以在 ./include/net/sch_generic.h 中找到：
static inline bool qdisc_run_begin(struct Qdisc *qdisc) { if (qdisc_is_running(qdisc)) return false; qdisc->__state |= __QDISC___STATE_RUNNING; return true; }
这个函数很简单：检查 qdisc 的 __state 标志。 如果它已经在运行，则返回 false。 否则，更新 __state 以启用 __QDISC___STATE_RUNNING 位。
同样，qdisc_run_end也是寡淡的：
static inline void qdisc_run_end(struct Qdisc *qdisc) { qdisc->__state &= ~__QDISC___STATE_RUNNING; }
它只是禁用 qdisc __state 字段中的 __QDISC__STATE_RUNNING 位。 需要注意的是，这两个函数都只是翻转位；自己既不实际开始，也不停止处理。 另一方面，函数 __qdisc_run 实际上开始处理。
__qdisc_run
__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); }
该函数首先获取 weight_p 值。 该值通常是 sysctl 设置的，也会在接收路径中使用。我们稍后会看到如何调整这个值。 这个循环做两件事：
- 它在一个繁忙的循环中调用
qdisc_restart，直到返回 false（或者触发下面的 break）。 - 确定配额是否降至零以下或
need_resched()返回 true。 如果其中一个为true，则调用__netif_schedule并中断循环。
记住：到现在为止，内核仍然在执行代表用户程序对 sendmsg 的原始调用；用户程序当前正在累积系统时间。 如果用户程序已经用完了内核中的时间配额，那么 need_resched 将返回 true。 如果仍然有可用的配额，并且用户程序尚未使用完其时间片，qdisc_restart 将再次被调用。
让我们看看 qdisc_restart(q) 是如何工作的，然后我们将深入研究 __netif_schedule(q)。
qdisc_restart
让我们跳到 qdisc_restart 的代码中：
/* * NOTE: Called under qdisc_lock(q) with locally disabled BH. * * __QDISC_STATE_RUNNING guarantees only one CPU can process * this qdisc at a time. qdisc_lock(q) serializes queue accesses for * this queue. * * netif_tx_lock serializes accesses to device driver. * * qdisc_lock(q) and netif_tx_lock are mutually exclusive, * if one is grabbed, another must be free. * * Note, that this procedure can be called by a watchdog timer * * Returns to the caller: * 0 - queue is empty or throttled. * >0 - queue is not empty. * */ static inline int qdisc_restart(struct Qdisc *q) { struct netdev_queue *txq; struct net_device *dev; spinlock_t *root_lock; struct sk_buff *skb; /* Dequeue packet */ skb = dequeue_skb(q); if (unlikely(!skb)) return 0; WARN_ON_ONCE(skb_dst_is_noref(skb)); root_lock = qdisc_lock(q); dev = qdisc_dev(q); txq = netdev_get_tx_queue(dev, skb_get_queue_mapping(skb)); return sch_direct_xmit(skb, q, dev, txq, root_lock); }
qdisc_restart 函数以一个有用的注释开始，该注释描述了调用此函数的一些加锁约束。 此函数执行的第一个操作是尝试从 qdisc 出队 skb。
函数 dequeue_skb 尝试获得下一个要传输的数据包。 如果队列为空 qdisc_restart 将返回 false（导致 __qdisc_run 退出）。
假设存在要传输的数据，则代码继续获取 qdisc 队列锁、qdisc 的关联设备和传输队列的引用。
所有这些都会传递到 sch_direct_xmit。 让我们先看一下 dequeue_skb，然后再看 sch_direct_xmit。
dequeue_skb
让我们看一下 ./net/sched/sch_generic.c 中的 dequeue_skb。 此函数处理两种主要情况：
- 将之前无法发送而重新排队的数据出队，或
- 将要处理的新数据从 qdisc 出队。
我们来看一下第一个案例：
static inline struct sk_buff *dequeue_skb(struct Qdisc *q) { struct sk_buff *skb = q->gso_skb; const struct netdev_queue *txq = q->dev_queue; if (unlikely(skb)) { /* check the reason of requeuing without tx lock first */ txq = netdev_get_tx_queue(txq->dev, skb_get_queue_mapping(skb)); if (!netif_xmit_frozen_or_stopped(txq)) { q->gso_skb = NULL; q->q.qlen--; } else skb = NULL;
请注意，该代码首先引用 qdisc 的 gso_skb 字段。 此字段保存重新排队的数据的引用。 如果未重新排队数据，则此字段将为 NULL。 如果该字段不为 NULL，则代码继续获取数据的传输队列并检查队列是否停止。 如果队列没有停止，则清除 gso_skb 字段，并且减少队列长度计数器。 如果队列停止，数据仍然关联到 gso_skb，但此函数将返回 NULL。
让我们检查下一个案例，其中没有重新排队的数据：
} else { if (!(q->flags & TCQ_F_ONETXQUEUE) || !netif_xmit_frozen_or_stopped(txq)) skb = q->dequeue(q); } return skb; }
在没有数据被重新排队的情况下，另一个复杂的复合 if 语句被求值。 如果：
- qdisc 没有单个传输队列，或者
- 传输队列未停止
然后，调用 qdisc 的 dequeue 函数以获取新数据。 dequeue 的内部实现根据 qdisc 的实现和特性而有所不同。
该函数以返回待处理的数据结束。
sch_direct_xmit
现在我们来看看 sch_direct_xmit（在 ./net/sched/sch_generic.c 中），它是向下移动数据到网络设备的重要参与者。 让我们一点一点地来看看：
/* * Transmit one skb, and handle the return status as required. Holding the * __QDISC_STATE_RUNNING bit guarantees that only one CPU can execute this * function. * * Returns to the caller: * 0 - queue is empty or throttled. * >0 - queue is not empty. */ int sch_direct_xmit(struct sk_buff *skb, struct Qdisc *q, struct net_device *dev, struct netdev_queue *txq, spinlock_t *root_lock) { int ret = NETDEV_TX_BUSY; /* And release qdisc */ spin_unlock(root_lock); HARD_TX_LOCK(dev, txq, smp_processor_id()); if (!netif_xmit_frozen_or_stopped(txq)) ret = dev_hard_start_xmit(skb, dev, txq); HARD_TX_UNLOCK(dev, txq);
该代码首先释放 qdisc 锁，然后锁定传输锁。 注意，HARD_TX_LOCK 是一个宏：
#define HARD_TX_LOCK(dev, txq, cpu) { \ if ((dev->features & NETIF_F_LLTX) == 0) { \ __netif_tx_lock(txq, cpu); \ } \ }
此宏检查设备功能标志中是否设置了 NETIF_F_LLTX 标志。 此标志已弃用，新设备驱动程序不应使用此标志。 此内核版本中的大多数驱动程序都不使用此标志，因此此检查将评估为 true，并将获得此数据的传输队列的锁。
接下来，检查传输队列以确保它没有停止，然后调用 dev_hard_start_xmit。 我们将在后面看到，dev_hard_start_xmit 从 Linux 内核的网络设备子系统转换网络数据到设备驱动程序本身以进行传输。 存储此函数的返回码，然后检查该返回码以确定传输是否成功。
一旦这已经运行（或者由于队列停止而被跳过），则释放队列的传输锁。 让我们继续：
spin_lock(root_lock); if (dev_xmit_complete(ret)) { /* Driver sent out skb successfully or skb was consumed */ ret = qdisc_qlen(q); } else if (ret == NETDEV_TX_LOCKED) { /* Driver try lock failed */ ret = handle_dev_cpu_collision(skb, txq, q);
接下来，再次获取此 qdisc 的锁，然后检查 dev_hard_start_xmit。 第一种情况是调用 dev_xmit_complete 检查，它只是检查返回值以确定数据是否成功发送。 如果是，则设置 qdisc 队列长度为返回值。
如果 dev_xmit_complete 返回 false，则将检查返回值以查看 dev_hard_start_xmit 是否从设备驱动程序返回 NETDEV_TX_LOCKED。 当驱动程序尝试自己锁定传输队列并失败时，具有不推荐使用的 NETIF_F_LLTX 功能标志的设备可以返回 NETDEV_TX_LOCKED。 在这种情况下，调用 handle_dev_cpu_collision 来处理锁竞争。 我们稍后会仔细研究 handle_dev_cpu_collision，但现在，让我们继续 sch_direct_xmit 并查看捕获所有的分支：
} else { /* Driver returned NETDEV_TX_BUSY - requeue skb */ if (unlikely(ret != NETDEV_TX_BUSY)) net_warn_ratelimited("BUG %s code %d qlen %d\n", dev->name, ret, q->q.qlen); ret = dev_requeue_skb(skb, q); }
因此，如果驱动程序没有传输数据，并且传输锁未被持有，则可能是由于 NETDEV_TX_BUSY （如果没有打印警告）。NETDEV_TX_BUSY 可以由驱动程序返回，以指示设备或驱动程序“忙碌”并且现在不能传输数据。 在本例中，调用 dev_requeue_skb 将要重试的数据重新入队。
该函数（可能）调整返回值来结束：
if (ret && netif_xmit_frozen_or_stopped(txq)) ret = 0; return ret;
让我们深入了解 handle_dev_cpu_collision 和 dev_requeue_skb。
