linux内核IMQ源码实现分析.doc

本文档的Copyleft归wwwlkk所有，使用GPL发布，可以自由拷贝、转载，转载时请保持文档的完整性，严禁用于任何商业用途。 E-mail: 来源: /?business&aid=6&un=wwwlkk#7linux2.6.35内核IMQ源码实现分析（1）数据包截留并重新注入协议栈技术1（2）及时处理数据包技术2（3）IMQ设备数据包重新注入协议栈流程4（4）IMQ截留数据包流程4（5）IMQ在软中断中及时将数据包重新注入协议栈7（6）结束语9前言：IMQ用于入口流量整形和全局的流量控制，IMQ的配置是很简单的，但很少人分析过IMQ的内核实现，网络上也没有IMQ的源码分析文档，为了搞清楚IMQ的性能，稳定性，以及借鉴IMQ的技术，本文分析了IMQ的内核实现机制。首先揭示IMQ的核心技术：1. 如何从协议栈中截留数据包，并能把数据包重新注入协议栈。2. 如何做到及时的将数据包重新注入协议栈。实际上linux的标准内核已经解决了以上2个技术难点，第1个技术可以在NF_QUEUE机制中看到，第二个技术可以在发包软中断中看到。下面先介绍这2个技术。（1）数据包截留并重新注入协议栈技术okfn参数：int ip_rcv(struct sk_buff *skb, struct net_device *dev, struct packet_type *pt, struct net_device *orig_dev)。return NF_HOOK(NFPROTO_IPV4, NF_INET_PRE_ROUTING, skb, dev, NULL, ip_rcv_finish);/下一流程是进入函数ip_rcv_finish。static inline intNF_HOOK(uint8_t pf, unsigned int hook, struct sk_buff *skb,struct net_device *in, struct net_device *out,int (*okfn)(struct sk_buff *)/将ip_rcv_finish地址作为参数传入。return NF_HOOK_THRESH(pf, hook, skb, in, out, okfn, INT_MIN);static inline intNF_HOOK_THRESH(uint8_t pf, unsigned int hook, struct sk_buff *skb, struct net_device *in, struct net_device *out, int (*okfn)(struct sk_buff *), int thresh)int ret = nf_hook_thresh(pf, hook, skb, in, out, okfn, thresh);if (ret = 1)ret = okfn(skb);/根据函数地址进入之前的ip_rcv_finish函数。return ret;okfn应该是下一个流程函数，不应该是本流程函数，这样做就很稳定了。IMQ设备的做法就是使用到了这个okfn，IMQ设备发送数据包不是调用网卡驱动，而是根据okfn，将数据包发给下一个流程函数。这里函数的概念和模块的概念是一样的，一个函数就是一个模块，一个模块处理完就将数据扔给下一个模块处理，各个模块间的数据处理是互补干扰的。从这个概念看使用okfn是很合理的。（2）及时处理数据包技术QoS有个技术难点：将数据包入队，然后发送队列中合适的数据包，那么如何做到队列中的数据包被及时的发送出去呢？linux有一套机制来解决这个问题，下面先了解这套机制：int dev_queue_xmit(struct sk_buff *skb)。txq = dev_pick_tx(dev, skb);q = rcu_dereference_bh(txq-qdisc);获得设备上的发送队列。if (q-enqueue) rc = _dev_xmit_skb(skb, q, dev, txq);goto out;数据包入队后，整个入队流程就结束了以上是入队过程，那么需要保证入队的数据包被及时的发送出去。static inline int _dev_xmit_skb(struct sk_buff *skb, struct Qdisc *q, struct net_device *dev, struct netdev_queue *txq)spinlock_t *root_lock = qdisc_lock(q);int rc;spin_lock(root_lock);if (unlikely(test_bit(_QDISC_STATE_DEACTIVATED, &q-state) kfree_skb(skb);如果这个队列是未运行的，那么释放这个数据包rc = NET_XMIT_DROP; else if (q-flags & TCQ_F_CAN_BYPASS) & !qdisc_qlen(q) & !test_and_set_bit(_QDISC_STATE_RUNNING, &q-state) 如果已经有CPU在运行这个队列，那么字节返回，因为一个队列只能由一个CPU运行。/* * This is a work-conserving queue; there are no old skbs * waiting to be sent out; and the qdisc is not running - * xmit the skb directly. */如果一个队列是位运行的，说明这个队列里面没有数据包，此时可以直接发送这个包if (!(dev-priv_flags & IFF_XMIT_DST_RELEASE)skb_dst_force(skb);_qdisc_update_bstats(q, skb-len);if (sch_direct_xmit(skb, q, dev, txq, root_lock)试图直接发送数据包，如果没有发送成功，或者队列中还有待发数据包，返回值会大于0，那么，此时需要激活这个队列。_qdisc_run(q);激活运行队列elseclear_bit(_QDISC_STATE_RUNNING, &q-state);rc = NET_XMIT_SUCCESS; else skb_dst_force(skb);rc = qdisc_enqueue_root(skb, q);qdisc_run(q);spin_unlock(root_lock);return rc;上面看到了，发送队列被激活了，只要队列中有数据包待发送，队列总是处于激活状态，那么激活状态的队列是否能保证队列中的数据包被及时的发送吗？接下来看一下，激活状态的队列的特点。void _qdisc_run(struct Qdisc *q)unsigned long start_time = jiffies;while (qdisc_restart(q) 队列会一直发送数据包，/* * Postpone processing if * 1. another process needs the CPU; * 2. weve been doing it for too long. */if (need_resched() | jiffies != start_time) 如果占用的CPU的时间太长了，有其他的路径需要占用CPU，此时暂时停止队列的发包。_netif_schedule(q); 将队列加入发送软中断NET_TX_SOFTIRQ的处理队列，当软中断被执行时，队列又会继续发送数据包。break;clear_bit(_QDISC_STATE_RUNNING, &q-state);由于软中断被激活，软中断的优先级仅次于硬中断，这样就保证了队列会被及时的运行，即保证了数据包会被及时的发送。这是linux内核发送软中断的机制，IMQ就是利用了这个机制，不同点在于：正常的发送队列是将数据包发送给网卡驱动，而IMQ队列是将数据包发送给okfn函数。以上2个技术点就是IMQ的关键技术，下面是IMQ的具体流程。（3）IMQ设备数据包重新注入协议栈流程IMQ网络设备：static const struct net_device_ops imq_netdev_ops = .ndo_open= imq_open, .ndo_stop= imq_close, .ndo_start_xmit= imq_dev_xmit, .ndo_get_stats= imq_get_stats, ;static netdev_tx_t imq_dev_xmit(struct sk_buff *skb, struct net_device *dev)-nf_reinject(entry, NF_ACCEPT); - struct sk_buff *skb = entry-skb; - entry-okfn(skb);对IMQ网络设备来说，发送一个数据包，实际上就是将数据包重新注入到协议栈中。（4）IMQ截留数据包流程那么协议栈如何将数据包注入到IMQ设备中：static unsigned int imq_nf_hook(unsigned int hook, struct sk_buff *pskb,const struct net_device *indev,const struct net_device *outdev,int (*okfn)(struct sk_buff *)return (pskb-imq_flags & IMQ_F_ENQUEUE) ? NF_IMQ_QUEUE : NF_ACCEPT;static struct nf_hook_ops imq_ops = /* imq_ingress_ipv4 */.hook= imq_nf_hook,.owner= THIS_MODULE,.pf= PF_INET,.hooknum= NF_INET_PRE_ROUTING,#if defined(CONFIG_IMQ_BEHAVIOR_BA) | defined(CONFIG_IMQ_BEHAVIOR_BB).priority= NF_IP_PRI_MANGLE + 1,#else.priority= NF_IP_PRI_NAT_DST + 1,#endif,/* imq_egress_ipv4 */.hook= imq_nf_hook,.owner= THIS_MODULE,.pf= PF_INET,.hooknum= NF_INET_POST_ROUTING,#if defined(CONFIG_IMQ_BEHAVIOR_AA) | defined(CONFIG_IMQ_BEHAVIOR_BA).priority= NF_IP_PRI_LAST,#else.priority= NF_IP_PRI_NAT_SRC - 1,#endif,imq_init_module(void) err = imq_init_hooks();static int _init imq_init_hooks(void)int ret;nf_register_queue_imq_handler(&imq_nfqh);ret = nf_register_hooks(imq_ops, ARRAY_SIZE(imq_ops);if (ret NF_VERDICT_BITS)goto next_hook;#endifrcu_read_unlock();return ret;int nf_imq_queue(struct sk_buff *skb, struct list_head *elem, u_int8_t pf, unsigned int hook, struct net_device *indev, struct net_device *outdev, int (*okfn)(struct sk_buff *), unsigned int queuenum)return _nf_queue(skb, elem, pf, hook, indev, outdev, okfn, queuenum, true);这里又要看一下nf_queue的调用流程：static int _nf_queue(struct sk_buff *skb, struct list_head *elem, u_int8_t pf, unsigned int hook, struct net_device *indev, struct net_device *outdev, int (*okfn)(struct sk_buff *), unsigned int queuenum, bool imq_queue).省略.#if defined(CONFIG_IMQ) | defined(CONFIG_IMQ_MODULE)if (imq_queue) #if defined(CONFIG_IMQ) | defined(CONFIG_IMQ_MODULE)if (imq_queue)qh = rcu_dereference(queue_imq_handler);else#endifqh = rcu_dereference(queue_handlerpf);如果是TRUE则选择queue_imq_handlerqh = rcu_dereference(queue_imq_handler);或者IMQ入队句柄else#endifqh = rcu_dereference(queue_handlerpf);。省略。entry = kmalloc(sizeof(*entry) + afinfo-route_key_size, GFP_ATOMIC);if (!entry)goto err_unlock;*entry = (struct nf_queue_entry) .skb= skb,.elem= list_entry(elem, struct nf_hook_ops, list),.pf= pf,.hook= hook,.indev= indev,.outdev= outdev,.okfn= okfn,这个是数据包在协议栈中的下一个处理函数地址;。省略。status = qh-outfn(entry, queuenum);将数据包入队。省略。到这里，可以看到最终的IMQ入队函数是：static int imq_nf_queue(struct nf_queue_entry *entry, unsigned queue_num)下面只给出关键步骤代码static int imq_nf_queue(struct nf_queue_entry *entry, unsigned queue_num)index = entry-skb-imq_flags & IMQ_F_IFMASK;获得IMQ设备号dev = imq_devs_cacheindex;获得设备结构txq = imq_select_queue(dev, skb);q = rcu_dereference(txq-qdisc);获得网络设备上的队列qdisc_enqueue_root(skb_shared, q);进行入队操作_netif_schedule(q);激活队列，在软中断中发送数据包，实际上时将数据包重新注入协议栈（5）IMQ在软中断中及时将数据包重新注入协议栈static inline void qdisc_run(struct Qdisc *q)if (!test_and_set_bit(_QDISC_STATE_RUNNING, &q-state)_qdisc_run(q);void _qdisc_run(struct Qdisc *q)unsigned long start_time = jiffies;while (qdisc_restart(q) 在这里发送一个数据包/* * Postpone processing if * 1. another process needs the CPU; * 2. weve been doing it for too long. */if (need_resched() | jiffies != start_time) _netif_schedule(q);break;clear_bit(_QDISC_STATE_RUNNING, &q-state);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 = qdis