Linux内核中流量控制.doc

本文档的Copyleft归yfydz所有，使用GPL发布，可以自由拷贝，转载，转载时请保持文档的完整性，严禁用于任何商业用途。msn: yfydz_来源：1. 前言linux内核中提供了流量控制的相关处理功能，相关代码在net/sched目录下；而应用层上的控制是通过iproute2软件包中的tc来实现，tc和sched的关系就好象iptables和netfilter的关系一样，一个是用户层接口，一个是具体实现，关于tc的使用方法可详将Linux Advanced Routing HOWTO，本文主要分析内核中的具体实现。流控包括几个部分: 流控算法, 通常在net/sched/sch_*.c中实现, 缺省的是FIFO, 是比较典型的黑盒模式, 对外只看到入队和出对两个操作; 流控结构的操作处理; 和用户空间的控制接口, 是通过rtnetlink实现的。以下内核代码版本为。2. 控制入口2.1 控制入口linux流控功能反映为网卡设备的属性，表明是网络最底层的处理部分, 已经和上层的网络协议栈无关了:/* include/linux/netdevice.h */struct net_device./* Cache line mostly used on queue transmit path (qdisc)*/* device queue lock */spinlock_tqueue_lock _cacheline_aligned_in_smp;/ 这是发送数据时的队列处理struct Qdisc*qdisc;/ 网卡停止时保存网卡活动时的队列处理方法struct Qdisc*qdisc_sleeping;/ 网卡处理的数据队列链表struct list_headqdisc_list;/ 最大队列长度unsigned longtx_queue_len;/* Max frames per queue allowed */* Partially transmitted GSO packet. */struct sk_buff*gso_skb;/* ingress path synchronizer */ 输入流控锁spinlock_tingress_lock;/ 这是对于接收数据时的队列处理struct Qdisc*qdisc_ingress;.2.1.2 输出流控数据发出流控处理时，上层的所有处理已经完成，数据包已经交到网卡设备进行发送，在数据发送时进行相关的流控处理网络数据的出口函数为dev_queue_xmit(); 如果是接收流控, 数据只是刚从网卡设备中收到, 还未交到网络上层处理, 不过网卡的输入流控不是必须的, 缺省情况下并不进行流控，输入流控入口函数为ing_filter()函数，该函数被skb_receive_skb()调用：/* net/core/dev.c */int dev_queue_xmit(struct sk_buff *skb)struct net_device *dev = skb-dev;struct Qdisc *q;int rc = -ENOMEM;./* Updates of qdisc are serialized by queue_lock. * The struct Qdisc which is pointed to by qdisc is now a * rcu structure - it may be accessed without acquiring * a lock (but the structure may be stale.) The freeing of the * qdisc will be deferred until its known that there are no * more references to it. * * If the qdisc has an enqueue function, we still need to * hold the queue_lock before calling it, since queue_lock * also serializes access to the device queue. */ 获取网卡的qdisc指针, 此出不需要锁, 是各个CPU的私有数据q = rcu_dereference(dev-qdisc);#ifdef CONFIG_NET_CLS_ACTskb-tc_verd = SET_TC_AT(skb-tc_verd,AT_EGRESS);#endif/ 如果队列输入非空, 将数据包入队/ 对于物理网卡设备, 缺省使用的是FIFO qdisc, 该成员函数非空, 只有逻辑网卡/ 才可能为空if (q-enqueue) /* Grab device queue */ 加锁spin_lock(&dev-queue_lock);/ 可以直接访问dev-qdisc了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);./ 出队操作static inline void qdisc_run(struct net_device *dev)if (!netif_queue_stopped(dev) & !test_and_set_bit(_LINK_STATE_QDISC_RUNNING, &dev-state)_qdisc_run(dev);/* net/sched/sch_generic.c */void _qdisc_run(struct net_device *dev)/ 如果是noop_qdisc流控, 实际是丢包if (unlikely(dev-qdisc = &noop_qdisc)goto out;while (qdisc_restart(dev) state);/* Kick device. Note, that this procedure can be called by a watchdog timer, so that we do not check dev-tbusy flag here. Returns: 0 - queue is empty. 0 - queue is not empty, but throttled. tbusy != 0. NOTE: Called under dev-queue_lock with locally disabled BH.*/static inline int qdisc_restart(struct net_device *dev)struct Qdisc *q = dev-qdisc;struct sk_buff *skb;/* Dequeue packet */ 数据包出队if (skb = dev-gso_skb) | (skb = q-dequeue(q) unsigned nolock = (dev-features & NETIF_F_LLTX);dev-gso_skb = NULL;.2.1.3 输入流控输入流控好象不是必须的,目前内核需要配置CONFIG_NET_CLS_ACT选项才起作用：/* net/core/dev.c */int netif_receive_skb(struct sk_buff *skb).#ifdef CONFIG_NET_CLS_ACTif (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.static int ing_filter(struct sk_buff *skb) struct Qdisc *q;struct net_device *dev = skb-dev;int result = TC_ACT_OK;/ 如果网卡设备有输入流控处理if (dev-qdisc_ingress) _u32 ttl = (_u32) G_TC_RTTL(skb-tc_verd);if (MAX_RED_LOOP %s)n,skb-input_dev-name, skb-dev-name);return TC_ACT_SHOT;/ 设置数据包的TC参数skb-tc_verd = SET_TC_RTTL(skb-tc_verd,ttl);skb-tc_verd = SET_TC_AT(skb-tc_verd,AT_INGRESS);spin_lock(&dev-ingress_lock);if (q = dev-qdisc_ingress) != NULL)/ 数据入队result = q-enqueue(skb, q);spin_unlock(&dev-ingress_lock);return result;2.2 初始化本节先跟踪一下网卡设备的qdisc指针是如何被赋值的, 缺省赋值为何值.在网卡设备的初始化函数register_netdevice()函数中调用dev_init_scheduler()函数对网卡设备的流控队列处理进行了初始化, 也就是说每个网络网卡设备的qdisc指针都不会是空的:/* net/sched/sch_gentric.c */void dev_init_scheduler(struct net_device *dev)qdisc_lock_tree(dev);/ 处理发出数据的qdisc是必须的,而处理输入数据的qdisc_ingress则不是必须的/ 缺省情况下的qdiscdev-qdisc = &noop_qdisc;dev-qdisc_sleeping = &noop_qdisc;INIT_LIST_HEAD(&dev-qdisc_list);qdisc_unlock_tree(dev);dev_watchdog_init(dev);当然noop_qdisc中的调度是不可用的, 只进行丢包处理；网卡在激活(dev_open)时会调用dev_activate()函数重新对qdisc指针赋值，但未对qdisc_ingress赋值:/* net/sched/sch_generic.c */void dev_activate(struct net_device *dev)/* No queueing discipline is attached to device; create default one i.e. pfifo_fast for devices, which need queueing and noqueue_qdisc for virtual interfaces */ 如果当前的qdisc_sleeping是noop_qdisc，重新找一个流控操作指针if (dev-qdisc_sleeping = &noop_qdisc) struct Qdisc *qdisc;/ 前提条件是发送队列长度非0, 这正常情况肯定满足的if (dev-tx_queue_len) / 对dev设备建立fifo处理, 只是缺省的网卡发送策略: FIFO, 先入先出qdisc = qdisc_create_dflt(dev, &pfifo_fast_ops);if (qdisc = NULL) printk(KERN_INFO %s: activation failedn, dev-name);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);/ 对qdisc_sleeping赋值dev-qdisc_sleeping = qdisc;write_unlock(&qdisc_tree_lock);/ 如果现在网线没插, 返回if (!netif_carrier_ok(dev)/* Delay activation until next carrier-on event */return;spin_lock_bh(&dev-queue_lock);/ 将网卡当前的qdisc赋值为qdisc_sleeping所指的qdiscrcu_assign_pointer(dev-qdisc, dev-qdisc_sleeping);if (dev-qdisc != &noqueue_qdisc) / 启动dev-trans_start = jiffies;dev_watchdog_up(dev);spin_unlock_bh(&dev-queue_lock);qdisc_sleeping用于保存在网卡起效时使用的qdisc, 因为在网卡down或网线拔除不可用时, 网卡设备的qdisc指针会指向noqueue_qdisc, 该qdisc操作就只是丢包, 这就是为什么在没插网线情况下也可以调用发包函数处理的原因, 结果就是直接丢包。/* net/sched/sch_generic.c */void dev_deactivate(struct net_device *dev)struct Qdisc *qdisc;spin_lock_bh(&dev-queue_lock);qdisc = dev-qdisc;/ 将网卡当前qdisc设置为noop_qdiscdev-qdisc = &noop_qdisc;/ 释放原来的qdiscqdisc_reset(qdisc);spin_unlock_bh(&dev-queue_lock);dev_watchdog_down(dev);/* Wait for outstanding dev_queue_xmit calls. */synchronize_rcu();/* Wait for outstanding qdisc_run calls. */while (test_bit(_LINK_STATE_QDISC_RUNNING, &dev-state)yield();if (dev-gso_skb) kfree_skb(dev-gso_skb);dev-gso_skb = NULL;3. 数据结构流控处理对外表现是一个黑盒，外部只能看到数据入队和出队，但内部队列是如何操作和管理外面是不知道的；另外处理队列处理外，流控还有一个调度器，该调度器将数据进行分类，然后对不同类型的数据采取不同的流控处理，所分的类型可能是多级的，形成一个树型的分类树。流控的基本数据结构是struct Qdisc(queueing discipline，直译是“排队纪律”，意译为“流控”)，这是内核中为数不多的以大写字母开头结构名称之一：/* include/net/sch_generic.h */struct Qdisc/ 入队操作int (*enqueue)(struct sk_buff *skb, struct Qdisc *dev);/ 出队操作struct sk_buff *(*dequeue)(struct Qdisc *dev);/ 标志unsignedflags;#define TCQ_F_BUILTIN1#define TCQ_F_THROTTLED2#define TCQ_F_INGRESS4intpadded;/ Qdisc的基本操作结构struct Qdisc_ops*ops;/ 句柄u32handle;u32parent;atomic_trefcnt;/ 数据包链表头struct sk_buff_headq;/ 网卡设备struct net_device*dev;struct list_headlist;/ 统计信息struct gnet_stats_basicbstats;struct gnet_stats_queueqstats;/ 速率估计struct gnet_stats_rate_estrate_est;/ 流控锁spinlock_t*stats_lock;struct rcu_head q_rcu;int(*reshape_fail)(struct sk_buff *skb,struct Qdisc *q);/* This field is deprecated, but it is still used by CBQ * and it will live until better solution will be invented. */ 父节点, 但基本已经被淘汰了struct Qdisc*_parent;流控队列的基本操作结构:struct Qdisc_ops/ 链表中的下一个struct Qdisc_ops*next;/ 类别操作结构struct Qdisc_class_ops*cl_ops;/ Qdisc的名称, 从数组大小看应该就是网卡名称charidIFNAMSIZ;/ 私有数据大小intpriv_size;/ 入队int (*enqueue)(struct sk_buff *, struct Qdisc *);/ 出队struct sk_buff *(*dequeue)(struct Qdisc *);/ 将数据包重新排队int (*requeue)(struct sk_buff *, struct Qdisc *);/ 丢弃unsigned int(*drop)(struct Qdisc *);/ 初始化int(*init)(struct Qdisc *, struct rtattr *arg);/ 复位为初始状态,释放缓冲,删除定时器,清空计数器void(*reset)(struct Qdisc *);/ 释放void(*destroy)(struct Qdisc *);/ 更改Qdisc参数int(*change)(struct Qdisc *, struct rtattr *arg);/ 输出int(*dump)(struct Qdisc *, struct sk_buff *);int(*dump_stats)(struct Qdisc *, struct gnet_dump *);struct module*owner;流控队列类别操作结构:struct Qdisc_class_ops/* Child qdisc manipulation */ 减子节点int(*graft)(struct Qdisc *, unsigned long cl,struct Qdisc *, struct Qdisc *);/ 增加子节点struct Qdisc *(*leaf)(struct Qdisc *, unsigned long cl);/* Class manipulation routines */ 获取, 增加使用计数unsigned long(*get)(struct Qdisc *, u32 classid);/ 释放, 减少使用计数void(*put)(struct Qdisc *, unsigned long);/ 改变int(*change)(struct Qdisc *, u32, u32,struct rtattr *, unsigned long *);/ 删除int(*delete)(struct Qdisc *, unsigned long);/ 遍历void(*walk)(struct Qdisc *, struct qdisc_walker * arg);/* Filter manipulation */struct tcf_proto *(*tcf_chain)(struct Qdisc *, unsigned long);/ tc捆绑unsigned long(*bind_tcf)(struct Qdisc *, unsigned long,u32 classid);/ tc解除void(*unbind_tcf)(struct Qdisc *, unsigned long);/* rtnetlink specific */ 输出int(*dump)(struct Qdisc *, unsigned long,struct sk_buff *skb, struct tcmsg*);int(*dump_stats)(struct Qdisc *, unsigned long,struct gnet_dump *);/ 流控速率控制表结构struct qdisc_rate_tablestruct tc_ratespec rate;u32data256;struct qdisc_rate_table *next;intrefcnt;4. 基本操作各种流控算法是通过流控操作结构实现，然后这些操作结构登记到内核的流控链表，在使用时可为网卡构造新的流控结构，将该结构和某种流控操作结构挂钩，这样就实现网卡采用某种策略发送数据进行流控，所有操作在用户空间都可通过tc程序设置。4.1 Qdisc的一些基本操作4.1.1 分配新的流控结构/* net/sched/sch_generic.c */ 分配新的Qdisc结构, Qdisc的操作结构由函数参数指定struct Qdisc *qdisc_alloc(struct net_device *dev, struct Qdisc_ops *ops)void *p;struct Qdisc *sch;unsigned int size;int err = -ENOBUFS;/* ensure that the Qdisc and the private data are 32-byte aligned */ Qdisc空间按32字节对齐size = QDISC_ALIGN(sizeof(*sch);/ 增加私有数据空间size += ops-priv_size + (QDISC_ALIGNTO - 1);p = kzalloc(size, GFP_KERNEL);if (!p)goto errout;/ 确保从缓冲区中的sch到缓冲区结束点空间是32字节对齐的sch = (struct Qdisc *) QDISC_ALIGN(unsigned long) p);/ 填充字节的数量sch-padded = (char *) sch - (char *) p;/ +-+/ |_|_|/ / | pad | |/ p sch / 初始化链表, 将用于挂接到dev的Qdisc链表INIT_LIST_HEAD(&sch-list);/ 初始化数据包链表skb_queue_head_init(&sch-q);/ Qdisc结构参数sch-ops = ops;sch-enqueue = ops-enqueue;sch-dequeue = ops-dequeue;sch-dev = dev;/ 网卡使用计数增加dev_hold(dev);sch-stats_lock = &dev-queue_lock;atomic_set(&sch-refcnt, 1);return sch;errout:return ERR_PTR(-err);struct Qdisc * qdisc_create_dflt(struct net_device *dev, struct Qdisc_ops *ops)struct Qdisc *sch;/ 分配Qdisc结构sch = qdisc_alloc(dev, ops);if (IS_ERR(sch)goto errout;/ 如果没有初始化函数或者初始化成功, 返回Qdisc结构if (!ops-init | ops-init(sch, NULL) = 0)return sch;/ 初始化失败, 释放Qdiscqdisc_destroy(sch);errout:return NULL;/* Under dev-queue_lock and BH! */ 调用Qdisc操作函数中的reset函数void qdisc_reset(struct Qdisc *qdisc)struct Qdisc_ops *ops = qdisc-ops;if (ops-reset)ops-reset(qdisc);/* this is the rcu callback function to clean up a qdisc when there * are no further references to it */ 真正释放Qdisc缓冲区static void _qdisc_destroy(struct rcu_head *head)struct Qdisc *qdisc = container_of(head, struct Qdisc, q_rcu);/ qdisc-padded就是缓冲区头的位置kfree(char *) qdisc - qdisc-padded);/* Under dev-queue_lock and BH! */ 释放Qdiscvoid qdisc_destroy(struct Qdisc *qdisc)struct Qdisc_ops *ops = qdisc-ops;/ 检查Qdisc的使用计数if (qdisc-flags & TCQ_F_BUILTIN | !atomic_dec_and_test(&qdisc-refcnt)return;/ 将Qdisc从网卡设备的Qdisc链表中断开list_del(&qdisc-list);#ifdef CONFIG_NET_ESTIMATORgen_kill_estimator(&qdisc-bstats, &qdisc-rate_est);#endif/ 复位操作if (ops-reset)ops-reset(qdisc);/ 内部释放操作if (ops-destroy)ops-destroy(qdisc);/ 减少操作结构的模块计数module_put(ops-owner);/ 减少网卡使用计数dev_put(qdisc-dev);/ 对每个CPU的数据进行具体空间释放call_rcu(&qdisc-q_rcu, _qdisc_destroy);/* include/net/sch_generic.h */ 将skb包添加到数据队列最后static inline int _qdisc_enqueue_tail(struct sk_buff *skb, struct Qdisc *sch, struct sk_buff_head *list)/ 将数据包连接到数据包链表尾_skb_queue_tail(list, skb);/ 更新统计信息/ 当前队列中数据包的数据长度增加sch-qstats.backlog += skb-len;/ Qdisc处理的数据包数字节数增加sch-bstats.bytes += skb-len;sch-bstats.packets+;return NET_XMIT_SUCCESS;static inline int qdisc_enqueue_tail(struct sk_buff *skb, struct Qdisc *sch)return _qdisc_enqueue_tail(skb, sch, &sch-q);/ 将队列头的数据包出队列static inline struct sk_buff *_qdisc_dequeue_head(struct Qdisc *sch, struct sk_buff_head *list)/ 从skb链表中头skb包出队列struct sk_buff *skb = _skb_dequeue(list);/ 减少当前数据队列数据长度计数if (likely(skb != NULL)sch-qstats.backlog -= skb-len;return skb;static inline struct sk_buff *qdisc_dequeue_head(struct Qdisc *sch)return _qdisc_dequeue_head(sch, &sch-q);/ 将队列尾的数据包出队列static inline struct sk_buff *_qdisc_dequeue_tail(struct Qdisc *sch, struct sk_buff_head *list)/ 从链表为提出数据包struct sk_buff *skb = _skb_dequeue_tail(list);if (likely(skb != NULL)sch-qstats.backlog -= skb-len;return skb;static inline struct sk_buff *qdisc_dequeue_tail(struct Qdisc *sch)return _qdisc_dequeue_tail(sch, &sch-q);/ 将数据包重新入队static inline int _qdisc_requeue(struct sk_buff *skb, struct Qdisc *sch, struct sk_buff_head *list)/ 添加到队列头_skb_queue_head(list, skb);/ 增加队列数据长度计数, 但不增加Qdisc处理的数据包数字节数sch-qstats.backlog += skb-len;sch-qstats.requeues+;return NET_XMIT_SUCCESS;static inline int qdisc_requeue(struct sk_buff *skb, struct Qdisc *sch)return _qdisc_requeue(skb, sch, &sch-q);/ 复位Qdisc队列static inline void _qdisc_reset_queue(struct Qdisc *sch, struct sk_buff_head *list)/* * We do not know the backlog in bytes of this list, it * is up to the caller to correct it */ 释放Qdisc当前数据包队列中的所有数据包skb_queue_purge(list);static inline void qdisc_reset_queue(struct Qdisc *sch)_qdisc_reset_queue(sch, &sch-q);sch-qstats.backlog = 0;/ 丢弃Qdisc数据队列尾的数据包static inline unsigned int _qdisc_queue_drop(struct Qdisc *sch, struct sk_buff_head *list)/ 取队列尾数据包struct sk_buff *skb = _qdisc_dequeue_tail(sch, list);if (likely(skb != NULL) / 释放该数据包unsigned int len = skb-len;kfree_skb(skb);return len;return 0;static inline unsigned int qdisc_queue_drop(struct Qdisc *sch)return _qdisc_queue_drop(sch, &sch-q);/ 丢弃数据包static inline int qdisc_drop(struct sk_buff *skb, struct Qdisc *sch)/ 释放数据包kfree_skb(skb);/ 丢包计数增加sch-qstats.drops+;return NET_XMIT_DROP;/ 整形失败丢包static inline int qdisc_reshape_fail(struct sk_buff *skb, struct Qdisc *sch)sch-qstats.drops+;#ifdef CONFIG_NET_CLS_POLICEif (sch-reshape_fail = NULL | sch-reshape_fail(skb, sch)goto drop;return NET_XMIT_SUCCESS;drop:#endifkfree_skb(skb);return NET_XMIT_DROP;/* net/sched/sch_api.c */* We know handle. Find qdisc among all qdiscs attached to device (root qdisc, all its children, children of children etc.)*/ 根据句柄查找Qdisc, 句柄是个32位整数用于标识Qdisc的struct Qdisc *qdisc_lookup(struct net_device *dev, u32 handle)struct Qdisc *q;read_lock(&qdisc_tree_lock);/ 遍历dev设备Qdisc链表list_for_each_entry(q, &dev-qdisc_list, list) / 句柄相同,返回Qdiscif (q-handle = handle) read_unlock(&qdisc_tree_lock);return q;read_unlock(&qdisc_tree_lock);return NULL;/ 返回指定类别的Qdisc叶节点static struct Qdisc *qdisc_leaf(struct Qdisc *p, u32 classid)unsigned long cl;struct Qdisc *leaf;/ Qdisc类别操作struct Qdisc_class_ops *cops = p-ops-cl_ops;if (cops = NULL)return NULL;/ 获取指定classid类型的类别句柄cl = cops-get(p, classid);if (cl = 0)return NULL;/ 调用类别操作结构的left成员函数获取叶Qdisc节点leaf = cops-leaf(p, cl);cops-put(p, cl