Linux - QOS实现框架分析

Linux QOS实现框架分析

作者：徐晓东(joyxxd@163.com)

Linux中的QOS分为入口(Ingress)部分和出口(Egress)部分，入口部分主要用于进行入口流量限速(policing)，出口部分的QOS用于队列调度(queuing scheduling)。

以下分析所参考的linux内核版本为2.6.21。 1. Ingress QOS

Ingress QOS在内核的入口点有两个，但是不能同时启用，这取决于内核编译选项。当打开了

CONFIG_NET_CLS_ACT时，入口点在src/net/core/dev.c的netif_receive_skb函数中，代码片段如下： #ifdef CONFIG_NET_CLS_ACT if (pt_prev) {

ret = deliver_skb(skb, pt_prev, orig_dev);

pt_prev = NULL; /* noone else should process this after*/ } else {

skb->tc_verd = SET_TC_OK2MUNGE(skb->tc_verd); }

ret = ing_filter(skb);

if (ret == TC_ACT_SHOT || (ret == TC_ACT_STOLEN)) { kfree_skb(skb); goto out; }

skb->tc_verd = 0; ncls: #endif

进入ing_filter后，代码片段如下： if ((q = dev->qdisc_ingress) != NULL) result = q->enqueue(skb, q);

我们可以看到，在这里判断了是否在设备上配置了ingress调度规则，如果配置了则调用其enqueue函数进行处理。在这里实际上调用的是src/net/core/sched/sch_ingress.c中的ingress_enqueue函数。

当没有打开CONFIG_NET_CLS_ACT，而是打开了CONFIG_NET_CLS_POLICE和CONFIG_NETFILTER时，就会在netfilter的PREROUTING钩子点处调用ing_hook函数，该函数的代码片段如下： if (dev->qdisc_ingress) {

spin_lock(&dev->queue_lock);

if ((q = dev->qdisc_ingress) != NULL) fwres = q->enqueue(skb, q); spin_unlock(&dev->queue_lock); }

可以看出与ing_filter的处理部分类似，最终都是调用了ingress qdisc的enqueue函数（即ingress_enqueue函数）。 ing_hook函数是在src/net/sched/sch_ingress.c文件中进行注册的，在sch_ingress文件中定义了一个结构实例ingress_qdisc_ops，如下所示： static struct Qdisc_ops ingress_qdisc_ops = { .next = NULL,

.cl_ops = &ingress_class_ops, .id = \

.priv_size = sizeof(struct ingress_qdisc_data), .enqueue = ingress_enqueue, .dequeue = ingress_dequeue, .requeue = ingress_requeue, .drop = ingress_drop, .init = ingress_init, .reset = ingress_reset, .destroy = ingress_destroy, .change = NULL,

.dump = ingress_dump, .owner = THIS_MODULE, };

所有的qdisc都会有这样的一个对象实例，在模块初始化时会调用register_qdisc函数将自己的struct Qdisc_ops结构实例注册到链表中，该链表头是qdisc_base，定义在sch_api.c文件中，static struct Qdisc_ops *qdisc_base。 当通过tc qdisc命令配置了ingress qdisc规则时，会调用到ingress_init函数，进入ingress_init函数代码片段如下：

#ifndef CONFIG_NET_CLS_ACT #ifdef CONFIG_NETFILTER if (!nf_registered) {

if (nf_register_hook(&ing_ops) < 0) {

printk(\ return -EINVAL; }

nf_registered++;

if (nf_register_hook(&ing6_ops) < 0) {

printk(\ \ } else

nf_registered++; }

#endif #endif

我们看到当没有定义CONFIG_NET_CLS_ACT，但是定义了CONFIG_NETFILTER时，会调用

nf_register_hook函数注册ing_ops和ing6_ops结构实例，ing_ops和ing6_ops的定义如下： static struct nf_hook_ops ing_ops = { .hook = ing_hook, .owner = THIS_MODULE,

.pf = PF_INET,

.hooknum = NF_IP_PRE_ROUTING, .priority = NF_IP_PRI_FILTER + 1, };

static struct nf_hook_ops ing6_ops = { .hook = ing_hook, .owner = THIS_MODULE,

.pf = PF_INET6,

.hooknum = NF_IP6_PRE_ROUTING, .priority = NF_IP6_PRI_FILTER + 1, };

针对IPV4和IPV6协议，注册的hook函数都是ing_hook，hook点是在PREROUTING，优先级低于NF_IP_PRI_FILTER。

当前比较推荐第一种使用方法，即打开CONFIG_NET_CLS_ACT选项，在netif_receive_skb函数中进入ingress的处理流程。

下面我们再跟踪一下ingress的enqueue处理流程，进入ingress_enqueue函数，代码片段如下： result = tc_classify(skb, p->filter_list, &res); ……

在ingress_enqueue函数中最主要的就是调用tc_classify函数进行分类操作，进入tc_classify函数，代码片段如下：

for ( ; tp; tp = tp->next) {

if ((tp->protocol == protocol ||

tp->protocol == __constant_htons(ETH_P_ALL)) && (err = tp->classify(skb, tp, res)) >= 0) { ……

其中tp是一个指向struct tcf_proto类型的指针，struct tcf_proto中包含了与filter相关的参数和函数指针。tp->classify调用了与该filter规则相相关联的classify函数，如果我们使用的是FW类型的filter，那么对应的classify函数就是fw_classify，该函数定义在src/net/sched/cls_fw.c文

件中。以FW分类器为例，我们再进入fw_classify函数，代码片段如下： struct fw_head *head = (struct fw_head*)tp->root; struct fw_filter *f; int r;

u32 id = skb->mark;

if (head != NULL) { id &= head->mask;

for (f=head->ht[fw_hash(id)]; f; f=f->next) { if (f->id == id) { *res = f->res;

#ifdef CONFIG_NET_CLS_IND

if (!tcf_match_indev(skb, f->indev)) continue;

#endif /* CONFIG_NET_CLS_IND */

r = tcf_exts_exec(skb, &f->exts, res); if (r < 0) continue;

return r; } }

} else {

/* old method */

if (id && (TC_H_MAJ(id) == 0 || !(TC_H_MAJ(id^tp->q->handle)))) { res->classid = id; res->class = 0; return 0; } }

return -1;

如上面代码所示，当filter中有规则时，遍历规则表，寻找与skb->mark（由ebtables或ip(6)tables来配置）相匹配的表项，如果找到了则会进一步调用tcf_exts_exec函数对扩展的action进行处理。再进入tcf_exts_exec函数，代码片段如下： #ifdef CONFIG_NET_CLS_ACT if (exts->action)

return tcf_action_exec(skb, exts->action, res); #elif defined CONFIG_NET_CLS_POLICE if (exts->police)

return tcf_police(skb, exts->police); #endif

如上面代码所示，当定义了CONFIG_NET_CLS_ACT选项，并且存在扩展action时调用tcf_action_exec函数进行处理，当定义了CONFIG_NET_CLS_POLICE选项，并且存在police扩展时会调用tcf_police函数。实际上CONFIG_NET_CLS_POLICE是一套老的police机制，它假定了在ingress处理

中只能进行police操作。CONFIG_NET_CLS_ACT是一套新的机制，该机制称为action扩展，即在filter规则处理之后，进一步进行额外的action处理，而Traffic Policing就属于其中的一种，它对应的文件是src/net/sched/act_police.c。如果我们通过tc filter命令配置fw的规则时用action选项指定了police作为扩展action，那么调用tcf_action_exec函数后就会最终调用到tcf_police函数。下面举个例子说明扩展action的配置方法： tc filter add dev eth0 parent ffff: protocol all prio 1 handle 100 fw action police rate 1000Kbit burst 2000k drop

从上面的命令可以看出，直接在fw后面加上action关键字即可配置action扩展。

下面我们再看看ingress qdisc是如何配置到接口上的，在这里只关注内核的处理流程，不关注tc工具与内核如何进行通信。通过tc命令进行配置后，会通过netlink接口调用内核中已经注册的一系列配置函数。

tc qdisc和tc clsss配置命令对应的配置函数在src/net/sched/sch_api.c的pktsched_init函数中进行了初始化注册，该函数在linux系统初始化的时候会被调用到。代码片断如下： if (link_p) {

link_p[RTM_NEWQDISC-RTM_BASE].doit = tc_modify_qdisc; link_p[RTM_DELQDISC-RTM_BASE].doit = tc_get_qdisc; link_p[RTM_GETQDISC-RTM_BASE].doit = tc_get_qdisc; link_p[RTM_GETQDISC-RTM_BASE].dumpit = tc_dump_qdisc; link_p[RTM_NEWTCLASS-RTM_BASE].doit = tc_ctl_tclass; link_p[RTM_DELTCLASS-RTM_BASE].doit = tc_ctl_tclass; link_p[RTM_GETTCLASS-RTM_BASE].doit = tc_ctl_tclass; link_p[RTM_GETTCLASS-RTM_BASE].dumpit = tc_dump_tclass; }

通过以上注册的一系列函数，就可以完成tc qdisc和class的命令配置。

tc filter配置命令对应的配置函数在src/net/sched/cls_api.c的tc_filter_init函数中进行了初始化注册，该函数也会在系统初始化的时候被调用到。代码片段如下： if (link_p) {

link_p[RTM_NEWTFILTER-RTM_BASE].doit = tc_ctl_tfilter; link_p[RTM_DELTFILTER-RTM_BASE].doit = tc_ctl_tfilter; link_p[RTM_GETTFILTER-RTM_BASE].doit = tc_ctl_tfilter; link_p[RTM_GETTFILTER-RTM_BASE].dumpit = tc_dump_tfilter; }

通过以上注册的一系列函数，就可以完成tc filter的命令配置。

例如在控制台调用tc qdisc add dev eth0 handle ffff: ingress命令后，最终会调用到内核的tc_modify_qdisc函数，进入该函数后可以找到以下的代码片段： if (clid == TC_H_INGRESS)

q = qdisc_create(dev, tcm->tcm_parent, tca, &err); else

q = qdisc_create(dev, tcm->tcm_handle, tca, &err);

其中，qdisc_create函数用于创建struct Qdisc结构实例，在qdisc_create函数中进行了几个关键操作：

(1) 查找struct Qdisc_ops结构实例 struct Qdisc_ops *ops;

ops = qdisc_lookup_ops(kind); //用于查找与kind值相对应的struct Qdisc_ops的结构实例。

关于struct Qdisc_ops结构，在上文中已经大概描述过，他在模块初始化的时候通过调用

register_qdisc函数注册到链表中，并且每一个qidsc都会对应一个struct Qdisc_ops结构实例，但是并不是每一个qdisc都会对应一个struct Qdisc结构实例，只有那些已经配置在接口上的qdisc实例才会和struct Qdisc实例相对应。 (2) 分配struct Qdisc结构实例 struct Qdisc *sch;

sch = qdisc_alloc(dev, ops); //用于分配并初始化struct Qdisc结构实例

(3) 将新分配的struct Qdisc结构实例加入到设备qdisc链表中

list_add_tail(&sch->list, &dev->qdisc_list);

执行完qdisc_create函数后，会继续执行qdisc_graft函数，之后是dev_graft_qdisc函数，最后会让dev->qdisc_ingress指针指向刚才创建的struct Qdisc结构实例，这样基本上就完成了ingress qdisc的配置。

如前文所述，在打开了CONFIG_NET_CLS_ACT配置选项后，还必须通过tc filter命令配置polic扩展action，否则进入ingress处理流程后将什么都不会做。 2. Egress QOS

出口队列调度的入口点在src/net/core/dev.c的dev_queue_xmit函数中，代码片段如下： q = rcu_dereference(dev->qdisc); #ifdef CONFIG_NET_CLS_ACT

skb->tc_verd = SET_TC_AT(skb->tc_verd,AT_EGRESS); #endif

if (q->enqueue) {

/* Grab device queue */

spin_lock(&dev->queue_lock); q = dev->qdisc; if (q->enqueue) {

rc = q->enqueue(skb, q); qdisc_run(dev);

spin_unlock(&dev->queue_lock);

rc = rc == NET_XMIT_BYPASS ? NET_XMIT_SUCCESS : rc; goto out; }

spin_unlock(&dev->queue_lock); }

从上面的代码中可以看出，通过q = rcu_dereference(dev->qdisc)可以获取到设备上root qdisc的指针q(struct Qdisc *)，在下面的处理过程中并没有判断q是否为NULL，这就说明设备上一定会存在egress qdisc，这一点和ingress是不同的，一个设备上可以没有ingress qdisc，即dev-> qdisc_ingress指针一般是NULL，除非通过tc qdisc命令配置了ingress qdisc。

我们看一下默认情况下dev->qdisc是如何配置的，在注册网络设备时会调用register_netdevice

函数，在register_netdevice函数中又会调用dev_init_scheduler函数，在该函数中的代码片段如下： dev->qdisc = &noop_qdisc;

dev->qdisc_sleeping = &noop_qdisc; INIT_LIST_HEAD(&dev->qdisc_list);

新注册一个接口后，dev->qdisc指针指向noop_qdisc结构实例，这是一个特殊的qdisc，它什么也不做。当创建好设备，用ifconfig up命令把设备拉起后，会调用到内核的src/net/core/dev.c中的dev_open函数，在dev_open函数中又会调用到src/net/sched/sch_generic.c中的dev_activate函数，代码片段如下：

if (dev->qdisc_sleeping == &noop_qdisc) { struct Qdisc *qdisc;

if (dev->tx_queue_len) {

qdisc = qdisc_create_dflt(dev, &pfifo_fast_ops, TC_H_ROOT);

if (qdisc == NULL) {

printk(KERN_INFO \ return; }

write_lock(&qdisc_tree_lock);

list_add_tail(&qdisc->list, &dev->qdisc_list); write_unlock(&qdisc_tree_lock); } else {

qdisc = &noqueue_qdisc; }

write_lock(&qdisc_tree_lock); dev->qdisc_sleeping = qdisc; write_unlock(&qdisc_tree_lock); } ……

rcu_assign_pointer(dev->qdisc, dev->qdisc_sleeping);

从上面的代码可以看出，当把设备拉起时给设备配置的默认root qdisc为pfifo_fast。之后我们可以在控制台调用tc qdisc add……命令配置其他的qdisc，配置过程与配置ingress qdisc的过程类似，在这里就不再赘述了。

下面我们以prio qdisc为例，看一下出口队列调度的大概流程。

假设我们已经通过tc qdisc命令在接口上配置了prio qdisc作为root qdisc，那么在dev_queue_xmit函数中调用了rc = q->enqueue(skb, q)后，就会调用到与prio qdisc对应的enqueue函数。对于prio qdisc来说，对应的enqueue函数是prio_enqueue(src/net/sched/sch_prio.c), 在prio_enqueue函数中代码片段如下：

qdisc = prio_classify(skb, sch, &ret); //进行分类选择，找出子qdisc

/*调用子qdisc的enqueue函数*/

if ((ret = qdisc->enqueue(skb, qdisc)) == NET_XMIT_SUCCESS) { sch->bstats.bytes += skb->len; sch->bstats.packets++;

sch->q.qlen++;

return NET_XMIT_SUCCESS; }

在上面的代码中首先通过调用prio_classify函数查找子qdisc，然后再调用子qdisc对应的enqueue函数。在这里我们要补充一些概念，在Linux QOS中qidsc分为两种，一种是有分类(classful)的qdisc，另一种是无分类（classless）的qdisc，有分类和无分类的qdisc都可以做为设备的root qdisc，但是有分类的qdisc通过它的分类(class)又可以嫁接出子qdisc，子qdisc可以是有分类的qdisc也可以是无分类的qdisc，最后的叶子qdisc则必须是无分类的qdisc，一般常用pfifo/bfifo做为叶子qdisc。因此可以利用这种组合的特性利用有分类qdisc和无分类qdisc组合出复杂的调度方式。下图简单的描述了他们之间的关系。

从上图中可以看出，当需要多种qdisc进行组合时，子qdisc必须嫁接在其父qdisc的某一个class上。

继续描述上面的prio qdisc，我们已经将prio qdisc配置成了root qdisc，在配置prio qdisc的过程中需要制定bands参数，这个参数指明了prio qdisc上的class数目，默认是3个class。配置了prio qdisc后会调用prio_init函数，该函数中会调用prio_tune函数，然后在prio_tune函数中会为prio qidsc创建默认的子qdisc，代码片段如下： for (i=0; ibands; i++) {

if (q->queues[i] == &noop_qdisc) { struct Qdisc *child;

child = qdisc_create_dflt(sch->dev, &pfifo_qdisc_ops, TC_H_MAKE(sch->handle, i + 1)); if (child) {

sch_tree_lock(sch);

child = xchg(&q->queues[i], child);

if (child != &noop_qdisc) {

qdisc_tree_decrease_qlen(child, child->q.qlen);

qdisc_destroy(child); }

sch_tree_unlock(sch); } } }

从上面的代码可以看出，默认为prio qdisc创建的子qdisc为pfifo qdisc。

通过上面你的描述，我们了解了root qdisc和child qdisc之间的关系，下面我们继续描述prio_classify函数，代码片段如下所示：

struct prio_sched_data *q = qdisc_priv(sch); u32 band = skb->priority; struct tcf_result res;

*qerr = NET_XMIT_BYPASS;

if (TC_H_MAJ(skb->priority) != sch->handle) { #ifdef CONFIG_NET_CLS_ACT

switch (tc_classify(skb, q->filter_list, &res)) { case TC_ACT_STOLEN: case TC_ACT_QUEUED:

*qerr = NET_XMIT_SUCCESS; case TC_ACT_SHOT: return NULL; };

if (!q->filter_list ) { #else

if (!q->filter_list || tc_classify(skb, q->filter_list, &res)) { #endif

if (TC_H_MAJ(band)) band = 0;

return q->queues[q->prio2band[band&TC_PRIO_MAX]]; }

band = res.classid; }

band = TC_H_MIN(band) - 1; if (band > q->bands)

return q->queues[q->prio2band[0]];

return q->queues[band];

从上面的代码可以看出，首先会用skb->priority字段的高16位的值和sch->handle值进行比较，如果不相等的话就会调用tc_classify分类匹配函数进行处理。假如我们使用的是fw分类方法的话，就会调用到fw_classify函数，该函数的处理流程我们在前文中已经描述过了，在此不再赘述。

当找到子qdisc的指针后，就调用子qdisc的enqueue函数，如前文所述prio qidsc默认的子qdisc为pfifo qdisc，因此在这里会调用pfifo qdisc对应的enqueue函数pfifo_enqueue

（src/net/sched/sch_fifo.c）,进入pfifo_enqueue函数，代码片段如下： struct fifo_sched_data *q = qdisc_priv(sch);

if (likely(skb_queue_len(&sch->q) < q->limit)) return qdisc_enqueue_tail(skb, sch);

return qdisc_reshape_fail(skb, sch);

用当前的队列长度和队列总长度进行比较(按数据包数量进行比较)，如果没有超出队列长度限制则把数据包缓冲到队列的尾部，否则就调用qdisc_reshape_fail函数丢弃数据包。至此就完成了数据包的enqueue操作。

数据包入队操作完成后，接下来在dev_queue_xmit函数中会调用qdisc_run函数进行队列调度和出队列操作,在该函数中会调用__qdisc_run函数,在该函数中代码片段如下： while (qdisc_restart(dev) < 0 && !netif_queue_stopped(dev)) /* NOTHING */; out:

clear_bit(__LINK_STATE_QDISC_RUNNING, &dev->state);

可以看到在这里循环调用了qdisc_restart函数(src/net/sched/sch_generic.c)，直到函数返回值不为负值或设备状态处于队列停止状态为止。进入qdisc_restart函数，代码片段如下： if (((skb = dev->gso_skb)) || ((skb = q->dequeue(q)))) { …… {

spin_unlock(&dev->queue_lock);

if (!netif_queue_stopped(dev)) { int ret;

ret = dev_hard_start_xmit(skb, dev); if (ret == NETDEV_TX_OK) { if (!nolock) {

netif_tx_unlock(dev); }

spin_lock(&dev->queue_lock); return -1; }

if (ret == NETDEV_TX_LOCKED && nolock) { spin_lock(&dev->queue_lock); goto collision; } } ……

spin_lock(&dev->queue_lock); q = dev->qdisc; }

requeue:

if (skb->next)

dev->gso_skb = skb; else

q->ops->requeue(skb, q); netif_schedule(dev); return 1; }

BUG_ON((int) q->q.qlen < 0); return q->q.qlen;

在这里调用了dequeue函数做出队列操作，对于prio qdisc来说会调用prio_dequeue函数，该函数的代码片段如下：

struct sk_buff *skb;

struct prio_sched_data *q = qdisc_priv(sch); int prio;

struct Qdisc *qdisc;

for (prio = 0; prio < q->bands; prio++) { qdisc = q->queues[prio]; skb = qdisc->dequeue(qdisc); if (skb) {

sch->q.qlen--; return skb; } }

return NULL;

这里对prio qdisc的所有bands(class)进行遍历，由于高优先级的class对应的的prio值也较小，因此直接遍历数组即可对优先级调度进行控制。最后还是会调用到子qidsc的dequeue函数获取数据包指针。当获取到的skb非空时直接返回，下次重新调用该函数时又会从高优先级的class开始遍历，这样就保证了每一次的dequeue操作总是从高优先级的class开始。

dequeue操作完成后，会调用dev_hard_start_xmit函数将数据包发送出去,如果发送成功则最终返回-1,在__qdisc_run函数中通过循环又会进入到qdisc_restart函数，直到队列为空为止。我们注意到在特殊情况下可能会有调用dev_hard_start_xmit发包不成功的情况，如果是因为发送设备忙碌造成的发送不成功则会进入requeue流程，将数据包重新缓冲到队列里，然后调用netif_schedule函数启用网络发包软中断处理流程，在软中断处理流程中会调用net_tx_action函数，最终又会调用qdisc_run进入队列调度流程。

3. 总结

以上对Ingress和Egress QOS的实现流程和框架进行了粗略分析，通过以上分析，希望读者可以从大的流程上了解QOS的实现框架。由于内核中实现了多种qdisc，在此不能一一赘述，有兴趣的读者可以参考本文的分析流程，切入到自己感兴趣的部分进行深入分析即可。但是万变不离其中，不论是哪种qdisc，都不会超出本文所描述的框架和流程,只不过在队列调度算法上会比较复杂罢了。

if (skb->next)

dev->gso_skb = skb; else

q->ops->requeue(skb, q); netif_schedule(dev); return 1; }

BUG_ON((int) q->q.qlen < 0); return q->q.qlen;

在这里调用了dequeue函数做出队列操作，对于prio qdisc来说会调用prio_dequeue函数，该函数的代码片段如下：

struct sk_buff *skb;

struct prio_sched_data *q = qdisc_priv(sch); int prio;

struct Qdisc *qdisc;

for (prio = 0; prio < q->bands; prio++) { qdisc = q->queues[prio]; skb = qdisc->dequeue(qdisc); if (skb) {

sch->q.qlen--; return skb; } }

return NULL;

这里对prio qdisc的所有bands(class)进行遍历，由于高优先级的class对应的的prio值也较小，因此直接遍历数组即可对优先级调度进行控制。最后还是会调用到子qidsc的dequeue函数获取数据包指针。当获取到的skb非空时直接返回，下次重新调用该函数时又会从高优先级的class开始遍历，这样就保证了每一次的dequeue操作总是从高优先级的class开始。

dequeue操作完成后，会调用dev_hard_start_xmit函数将数据包发送出去,如果发送成功则最终返回-1,在__qdisc_run函数中通过循环又会进入到qdisc_restart函数，直到队列为空为止。我们注意到在特殊情况下可能会有调用dev_hard_start_xmit发包不成功的情况，如果是因为发送设备忙碌造成的发送不成功则会进入requeue流程，将数据包重新缓冲到队列里，然后调用netif_schedule函数启用网络发包软中断处理流程，在软中断处理流程中会调用net_tx_action函数，最终又会调用qdisc_run进入队列调度流程。

3. 总结

以上对Ingress和Egress QOS的实现流程和框架进行了粗略分析，通过以上分析，希望读者可以从大的流程上了解QOS的实现框架。由于内核中实现了多种qdisc，在此不能一一赘述，有兴趣的读者可以参考本文的分析流程，切入到自己感兴趣的部分进行深入分析即可。但是万变不离其中，不论是哪种qdisc，都不会超出本文所描述的框架和流程,只不过在队列调度算法上会比较复杂罢了。

本文来源：https://www.bwwdw.com/article/7oww.html