Linux网络之设备接口层:发送数据包流程dev_queue_xmit

时间：2023-12-30 17:07:56

转自：http://blog.csdn.net/wdscq1234/article/details/51926808

写在前面

本文主要是分析kernel-3.8的源代码，主要集中在Network的netdevice层面，来贯穿interface传输数据包的流程，kernel 博大精深，这也仅仅是一点个人愚见，作为一个笔记形式的文章，如有错误或者表述不当之处，还请大家留言批评指正，非常感谢！

主要涉及的file：kernel-3.18/net/core/dev.c

kernel-3.18/net/sched/sch_generic.c

dev_queue_xmit分析

当上层的APP试图建立一个TCP的链接，或者发送一个封包的时候，在kernel的协议栈部分，在TCP/UDP层会组成一个网络的封包，然后通过IP进行路由选择以及iptables的Hook，之后 到neighbor层查询或者询问下一跳的链路层地址，然后通过调用dev_queue_xmit这个网络设备接口层函数发送给driver，本文就来分析一下dev_queue_xmit的相关流程，了解一个包是如何发送出去的！

  1. int dev_queue_xmit(struct sk_buff *skb)
  2. {
  3. return __dev_queue_xmit(skb, NULL);
  4. }

这个是直接调用的__dev_queue_xmit 传入的参数是一个skb 数据包

需要确认的几个点就是

1.设备在调用这个函数之前，必须设置设备优先级 和缓冲区buffer

2.如果此函数发送失败，会返回一个负数的Error number，不过即使返回正数，也不一定保证发送成功，封包也许会被网络拥塞给drop掉

3.这个函数也可以从队列规则中返回error，NET_XMIT_DROP， 这个错误是一个整数，所以错误也有可能是整数，也验证了点2 ，所以在协议栈的上一层使用这个函数的时候，可能需要注意error的处理部分

4. 不管这个函数返回什么值，这个skb最终的宿命就是被consume，也就是free掉了...  所以这个时候上层不要尝试重发了... 除非有协议栈的重传， 要不skb已经被free了，再去重新去调用,不要命了..此时kernel就挂掉了...

5.在调用这个函数的时候，必须打开中断，这些东西不是特别明白原因....

*      When calling this method, interrupts MUST be enabled.  This is because
 *      the BH enable code must have IRQs enabled so that it will not deadlock.

  1. static int __dev_queue_xmit(struct sk_buff *skb, void *accel_priv)
  2. {
  3. struct net_device *dev = skb->dev;
  4. struct netdev_queue *txq;
  5. struct Qdisc *q;
  6. int rc = -ENOMEM;
  7. skb_reset_mac_header(skb);
  8. if (unlikely(skb_shinfo(skb)->tx_flags & SKBTX_SCHED_TSTAMP))
  9. __skb_tstamp_tx(skb, NULL, skb->sk, SCM_TSTAMP_SCHED);
  10. /* Disable soft irqs for various locks below. Also
  11. * stops preemption for RCU.
  12. */
  13. rcu_read_lock_bh();
  14. skb_update_prio(skb);
  15. /* If device/qdisc don't need skb->dst, release it right now while
  16. * its hot in this cpu cache.
  17. */
  18. /*这个地方看netdevcie的flag是否要去掉skb DST相关的信息，一般情况下这个flag是默认被设置的
  19. *在alloc_netdev_mqs的时候，已经默认给设置了，其实个人认为这个路由信息也没有太大作用了...
  20. *dev->priv_flags = IFF_XMIT_DST_RELEASE | IFF_XMIT_DST_RELEASE_PERM;
  21. */
  22. if (dev->priv_flags & IFF_XMIT_DST_RELEASE)
  23. skb_dst_drop(skb);
  24. else
  25. skb_dst_force(skb);
  26. /*此处主要是取出此netdevice的txq和txq的Qdisc,Qdisc主要用于进行拥塞处理，一般的情况下，直接将
  27. *数据包发送给driver了，如果遇到Busy的状况，就需要进行拥塞处理了，就会用到Qdisc*/
  28. txq = netdev_pick_tx(dev, skb, accel_priv);
  29. q = rcu_dereference_bh(txq->qdisc);
  30. #ifdef CONFIG_NET_CLS_ACT
  31. skb->tc_verd = SET_TC_AT(skb->tc_verd, AT_EGRESS);
  32. #endif
  33. trace_net_dev_queue(skb);
  34. /*如果Qdisc有对应的enqueue规则，就会调用__dev_xmit_skb，进入带有拥塞的控制的Flow，注意这个地方，虽然是走拥塞控制的
  35. *Flow但是并不一定非得进行enqueue操作啦，只有Busy的状况下，才会走Qdisc的enqueue/dequeue操作进行
  36. */
  37. if (q->enqueue) {
  38. rc = __dev_xmit_skb(skb, q, dev, txq);
  39. goto out;
  40. }
  41. /* The device has no queue. Common case for software devices:
  42. loopback, all the sorts of tunnels...
  43. Really, it is unlikely that netif_tx_lock protection is necessary
  44. here.  (f.e. loopback and IP tunnels are clean ignoring statistics
  45. counters.)
  46. However, it is possible, that they rely on protection
  47. made by us here.
  48. Check this and shot the lock. It is not prone from deadlocks.
  49. Either shot noqueue qdisc, it is even simpler 8)
  50. */
  51. /*此处是设备没有Qdisc的，实际上没有enqueue/dequeue的规则，无法进行拥塞控制的操作，
  52. *对于一些loopback/tunnel interface比较常见，判断下设备是否处于UP状态*/
  53. if (dev->flags & IFF_UP) {
  54. int cpu = smp_processor_id(); /* ok because BHs are off */
  55. if (txq->xmit_lock_owner != cpu) {
  56. if (__this_cpu_read(xmit_recursion) > RECURSION_LIMIT)
  57. goto recursion_alert;
  58. skb = validate_xmit_skb(skb, dev);
  59. if (!skb)
  60. goto drop;
  61. HARD_TX_LOCK(dev, txq, cpu);
  62. /*这个地方判断一下txq不是stop状态，那么就直接调用dev_hard_start_xmit函数来发送数据*/
  63. if (!netif_xmit_stopped(txq)) {
  64. __this_cpu_inc(xmit_recursion);
  65. skb = dev_hard_start_xmit(skb, dev, txq, &rc);
  66. __this_cpu_dec(xmit_recursion);
  67. if (dev_xmit_complete(rc)) {
  68. HARD_TX_UNLOCK(dev, txq);
  69. goto out;
  70. }
  71. }
  72. HARD_TX_UNLOCK(dev, txq);
  73. net_crit_ratelimited("Virtual device %s asks to queue packet!\n",
  74. dev->name);
  75. } else {
  76. /* Recursion is detected! It is possible,
  77. * unfortunately
  78. */
  79. recursion_alert:
  80. net_crit_ratelimited("Dead loop on virtual device %s, fix it urgently!\n",
  81. dev->name);
  82. }
  83. }
  84. rc = -ENETDOWN;
  85. drop:
  86. rcu_read_unlock_bh();
  87. atomic_long_inc(&dev->tx_dropped);
  88. kfree_skb_list(skb);
  89. return rc;
  90. out:
  91. rcu_read_unlock_bh();
  92. return rc;
  93. }

从对_dev_queue_xmit函数的分析来看，发送报文有2中情况：

1.有拥塞控制策略的情况，比较复杂，但是目前最常用

2.没有enqueue的状况，比较简单，直接发送到driver，如loopback等使用

先检查是否有enqueue的规则，如果有即调用__dev_xmit_skb进入拥塞控制的flow，如果没有且txq处于On的状态，那么就调用dev_hard_start_xmit直接发送到driver，好 那先分析带Qdisc策略的flow 进入__dev_xmit_skb

__dev_xmit_skb分析

  1. static inline int __dev_xmit_skb(struct sk_buff *skb, struct Qdisc *q,
  2. struct net_device *dev,
  3. struct netdev_queue *txq)
  4. {
  5. spinlock_t *root_lock = qdisc_lock(q);
  6. bool contended;
  7. int rc;
  8. qdisc_pkt_len_init(skb);
  9. qdisc_calculate_pkt_len(skb, q);
  10. /*
  11. * Heuristic to force contended enqueues to serialize on a
  12. * separate lock before trying to get qdisc main lock.
  13. * This permits __QDISC___STATE_RUNNING owner to get the lock more
  14. * often and dequeue packets faster.
  15. */
  16. contended = qdisc_is_running(q);
  17. if (unlikely(contended))
  18. spin_lock(&q->busylock);
  19. spin_lock(root_lock);
  20. /*这个地方主要是判定Qdisc的state: __QDISC_STATE_DEACTIVATED,如果处于非活动的状态，就DROP这个包，返回NET_XMIT_DROP
  21. *一般情况下带有Qdisc策略的interface，在被close的时候才会打上这个flag */
  22. if (unlikely(test_bit(__QDISC_STATE_DEACTIVATED, &q->state))) {
  23. kfree_skb(skb);
  24. rc = NET_XMIT_DROP;
  25. } else if ((q->flags & TCQ_F_CAN_BYPASS) && !qdisc_qlen(q) &&
  26. qdisc_run_begin(q)) {
  27. /*
  28. * This is a work-conserving queue; there are no old skbs
  29. * waiting to be sent out; and the qdisc is not running -
  30. * xmit the skb directly.
  31. */
  32. /* 结合注释以及code来看，此处必须满足3个调节才可以进来，
  33. * 1.flag必须有TCQ_F_CAN_BYPASS，默认条件下是有的，表明可以By PASS Qdisc规则
  34. * 2.q的len为0，也就是说Qdisc中一个包也没有
  35. * 3.Qdisc 起初并没有处于running的状态，然后置位Running！
  36. * 满足上述3个条件调用sch_direct_xmit
  37. */
  38. qdisc_bstats_update(q, skb);
  39. /*这个函数*/
  40. if (sch_direct_xmit(skb, q, dev, txq, root_lock, true)) {
  41. if (unlikely(contended)) {
  42. spin_unlock(&q->busylock);
  43. contended = false;
  44. }
  45. __qdisc_run(q);
  46. } else
  47. qdisc_run_end(q);
  48. rc = NET_XMIT_SUCCESS;
  49. } else {
  50. /*如果上述3个条件其中任何一个或者多个不满足，就要进行enqueue操作了，这个地方其实就是表明通讯出现拥塞，需要进行管理了
  51. *如果q不是运行状态，就设置成运行状况，如果一直是运行状态，那么就不用管了！*/
  52. rc = q->enqueue(skb, q) & NET_XMIT_MASK;
  53. if (qdisc_run_begin(q)) {
  54. if (unlikely(contended)) {
  55. spin_unlock(&q->busylock);
  56. contended = false;
  57. }
  58. __qdisc_run(q);
  59. }
  60. }
  61. spin_unlock(root_lock);
  62. if (unlikely(contended))
  63. spin_unlock(&q->busylock);
  64. return rc;
  65. }

从上述分析来看，可以分成2个状况：

1. Qdisc满足上述3个条件，置位TCQ_F_CAN_BYPASS,0个包，没有running直接调用sch_direct_xmit，感觉这种状况，是一开始刚发第一个包的时候肯定是这种状况...

2.不满足上述的3个条件 一个或者多个，那就直接进行enqueue操作，然后运行qdisc

个人认为，第一种状况是在网络通畅的状况下遇到的状况，qdisc的队列基本上处于空的状态，都是直接传送给driver了，第二种情况是属于出现网络拥塞的情况，出现发送失败的状况了

Q里面还有一些待发送的数据包，为了保证Q中的数据按照Qdisc的规则发送，比如先进先出，就需要enqueue操作，然后再去dequeue发送出去！

sch_direct_xmit

下面来分析sch_direct_xmit，这个函数可能传输几个数据包，因为在不经过queue状况下和经过queue的状况下都会调通过这个函数发送，如果是queue状况，肯定是能够传输多个数据包了，本文后面也有分析，并按照需求处理return的状态，需要拿着__QDISC___STATE_RUNNING  bit，只有一个CPU 可以执行这个函数， 在这里有可能会出现BUSY的状况！

  1. int sch_direct_xmit(struct sk_buff *skb, struct Qdisc *q,
  2. struct net_device *dev, struct netdev_queue *txq,
  3. spinlock_t *root_lock, bool validate)
  4. {
  5. int ret = NETDEV_TX_BUSY;
  6. /* And release qdisc */
  7. spin_unlock(root_lock);
  8. /* Note that we validate skb (GSO, checksum, ...) outside of locks */
  9. if (validate)
  10. skb = validate_xmit_skb_list(skb, dev);
  11. if (likely(skb)) {
  12. HARD_TX_LOCK(dev, txq, smp_processor_id());
  13. /*如果说txq被stop，即置位QUEUE_STATE_ANY_XOFF_OR_FROZEN，就直接ret = NETDEV_TX_BUSY
  14. *如果说txq 正常运行，那么直接调用dev_hard_start_xmit发送数据包*/
  15. if (!netif_xmit_frozen_or_stopped(txq))
  16. skb = dev_hard_start_xmit(skb, dev, txq, &ret);
  17. HARD_TX_UNLOCK(dev, txq);
  18. } else {
  19. spin_lock(root_lock);
  20. return qdisc_qlen(q);
  21. }
  22. spin_lock(root_lock);
  23. /*进行返回值处理！ 如果ret < NET_XMIT_MASK 为true 否则 flase*/
  24. if (dev_xmit_complete(ret)) {
  25. /* Driver sent out skb successfully or skb was consumed */
  26. /*这个地方需要注意可能有driver的负数的case，也意味着这个skb被drop了*/
  27. ret = qdisc_qlen(q);
  28. } else if (ret == NETDEV_TX_LOCKED) {
  29. /* Driver try lock failed */
  30. ret = handle_dev_cpu_collision(skb, txq, q);
  31. } else {
  32. /* Driver returned NETDEV_TX_BUSY - requeue skb */
  33. if (unlikely(ret != NETDEV_TX_BUSY))
  34. net_warn_ratelimited("BUG %s code %d qlen %d\n",
  35. dev->name, ret, q->q.qlen);
  36. /*发生Tx Busy的时候，重新进行requeue*/
  37. ret = dev_requeue_skb(skb, q);
  38. }
  39. /*如果txq stop并且ret ！=0  说明已经无法发送数据包了ret = 0*/
  40. if (ret && netif_xmit_frozen_or_stopped(txq))
  41. ret = 0;
  42. return ret;
  43. }

这个函数就是在txq没有被stop的状况下，直接发送给driver，如果遇到无法发送的状况，要么是fail的，要么出现Tx Busy就requeue，使用拥塞的方式进行发送。

dev_hard_start_xmit

继续看dev_hard_start_xmit，这个函数比较简单，调用xmit_one来发送一个到多个数据包了

  1. struct sk_buff *dev_hard_start_xmit(struct sk_buff *first, struct net_device *dev,
  2. struct netdev_queue *txq, int *ret)
  3. {
  4. struct sk_buff *skb = first;
  5. int rc = NETDEV_TX_OK;
  6. /*此处skb为什么会有链表呢？*/
  7. while (skb) {
  8. /*取出skb的下一个数据单元*/
  9. struct sk_buff *next = skb->next;
  10. /*置空，待发送数据包的next*/
  11. skb->next = NULL;
  12. /*将此数据包送到driver Tx函数，因为dequeue的数据也会从这里发送，所以会有netx！*/
  13. rc = xmit_one(skb, dev, txq, next != NULL);
  14. /*如果发送不成功，next还原到skb->next 退出*/
  15. if (unlikely(!dev_xmit_complete(rc))) {
  16. skb->next = next;
  17. goto out;
  18. }
  19. /*如果发送成功，把next置给skb，一般的next为空 这样就返回，如果不为空就继续发！*/
  20. skb = next;
  21. /*如果txq被stop，并且skb需要发送，就产生TX Busy的问题！*/
  22. if (netif_xmit_stopped(txq) && skb) {
  23. rc = NETDEV_TX_BUSY;
  24. break;
  25. }
  26. }
  27. out:
  28. *ret = rc;
  29. return skb;
  30. }

对于xmit_one这个来讲比较简单了，下面代码中列出了xmit_one, netdev_start_xmit,__netdev_start_xmit 这个三个函数，其目的就是将封包送到driver的tx函数了..中间在送往driver之前，还会经历抓包的过程，本文不介绍抓包的流程了。

  1. static int xmit_one(struct sk_buff *skb, struct net_device *dev,
  2. struct netdev_queue *txq, bool more)
  3. {
  4. unsigned int len;
  5. int rc;
  6. /*如果有抓包的工具的话，这个地方会进行抓包，such as Tcpdump*/
  7. if (!list_empty(&ptype_all))
  8. dev_queue_xmit_nit(skb, dev);
  9. len = skb->len;
  10. trace_net_dev_start_xmit(skb, dev);
  11. /*调用netdev_start_xmit，快到driver的tx函数了*/
  12. rc = netdev_start_xmit(skb, dev, txq, more);
  13. trace_net_dev_xmit(skb, rc, dev, len);
  14. return rc;
  15. }
  16. static inline netdev_tx_t netdev_start_xmit(struct sk_buff *skb, struct net_device *dev,
  17. struct netdev_queue *txq, bool more)
  18. {
  19. const struct net_device_ops *ops = dev->netdev_ops;
  20. int rc;
  21. /*__netdev_start_xmit 里面就完全是使用driver 的ops去发包了，其实到此为止，一个skb已经从netdevice
  22. *这个层面送到driver层了，接下来会等待driver的返回*/
  23. rc = __netdev_start_xmit(ops, skb, dev, more);
  24. /*如果返回NETDEV_TX_OK，那么会更新下Txq的trans时间戳哦，txq->trans_start = jiffies;*/
  25. if (rc == NETDEV_TX_OK)
  26. txq_trans_update(txq);
  27. return rc;
  28. }
  29. static inline netdev_tx_t __netdev_start_xmit(const struct net_device_ops *ops,
  30. struct sk_buff *skb, struct net_device *dev,
  31. bool more)
  32. {
  33. skb->xmit_more = more ? 1 : 0;
  34. return ops->ndo_start_xmit(skb, dev);
  35. }

分析到这里已经形成了一条线，__dev_xmit_skb中对于qlen为0的状态下直接调用sch_direct_xmit去发送，这个时候sch_direct_xmit也直接调用了dev_hard_start_xmit来直接送到driver，如果发送成功会返回NETDEV_TX_OK，如果由于一些原因产生TX_BUSY的话，重新将这个包requeu到Qdisc中，再使用拥塞的方式发送

拥塞发送方式分析

Qdisc简介

接下来再分析下另外一条线，如果不满足上述的3个条件，即interface配置不允许直接发送，或者是有发送失败的包或者积累的封包等，就需要被enqueue了，进入Qdisc的操作！

从code中看，直接是q->enqueue这样的钩子函数调用，那么这个函数在哪里被赋值的！

1. 注册的flow，默认是没有Qdisc的

register_netdevice-->dev_init_scheduler--->dev->qdisc = &noop_qdisc;  //默认是没有qdisc

可以看下Code中对noop_qdisc的规定...

  1. struct Qdisc noop_qdisc = {
  2. .enqueue    =   noop_enqueue,
  3. .dequeue    =   noop_dequeue,
  4. .flags      =   TCQ_F_BUILTIN,
  5. .ops        =   &noop_qdisc_ops,
  6. .list       =   LIST_HEAD_INIT(noop_qdisc.list),
  7. .q.lock     =   __SPIN_LOCK_UNLOCKED(noop_qdisc.q.lock),
  8. .dev_queue  =   &noop_netdev_queue,
  9. .busylock   =   __SPIN_LOCK_UNLOCKED(noop_qdisc.busylock),
  10. };
  1. static struct Qdisc_ops noqueue_qdisc_ops __read_mostly = {
  2. .id     =   "noqueue",
  3. .priv_size  =   0,
  4. .enqueue    =   noop_enqueue,
  5. .dequeue    =   noop_dequeue,
  6. .peek       =   noop_dequeue,
  7. .owner      =   THIS_MODULE,
  8. };

enqueue/dequeue规则 不能使用哈！

  1. /* "NOOP" scheduler: the best scheduler, recommended for all interfaces
  2. under all circumstances. It is difficult to invent anything faster or
  3. cheaper.
  4. */
  5. static int noop_enqueue(struct sk_buff *skb, struct Qdisc *qdisc)
  6. {
  7. kfree_skb(skb);
  8. return NET_XMIT_CN;
  9. }
  10. static struct sk_buff *noop_dequeue(struct Qdisc *qdisc)
  11. {
  12. return NULL;
  13. }

2.在打开设备的时候，赋予default

实际上刚开始注册的时候 就是这样 ，并没有任何可用的规则，但是在真正打开interface的时候，系统还是给赋予系统默认的！

dev_open-->__dev_open-->dev_activate

  1. /*如果没有Qdisc的规则设置到device，就给设备create一个默认的规则*/
  2. f (dev->qdisc == &noop_qdisc)
  3. attach_default_qdiscs(dev);
  1. static void attach_default_qdiscs(struct net_device *dev)
  2. {
  3. struct netdev_queue *txq;
  4. struct Qdisc *qdisc;
  5. txq = netdev_get_tx_queue(dev, 0);
  6. /*这个地方判断dev是否是多Q的设备，如果是不是多Q的设备，或者tx_queue_len为0 就会进入if判断
  7. *这个地方比较疑惑，if内部使用的是一个循环去遍历tx_queue 感觉是要处理multiqueue的状况，现在
  8. *却用来处理一个的...*/
  9. if (!netif_is_multiqueue(dev) || dev->tx_queue_len == 0) {
  10. netdev_for_each_tx_queue(dev, attach_one_default_qdisc, NULL);
  11. dev->qdisc = txq->qdisc_sleeping;
  12. atomic_inc(&dev->qdisc->refcnt);
  13. } else {
  14. /*如果不满足上述条件，就创建一个默认的，mq_qdisc_ops，从他的注释来看 mq 的Qdisc只使用与多Q状况*/
  15. qdisc = qdisc_create_dflt(txq, &mq_qdisc_ops, TC_H_ROOT);
  16. if (qdisc) {
  17. dev->qdisc = qdisc;
  18. qdisc->ops->attach(qdisc);
  19. }
  20. }
  21. }

我们在使用的netdevice基本上都是单队列的，默认情况下都是这个pfifo_fast_ops队列，具体Qdisc的知识这边不做深究！

  1. struct Qdisc_ops pfifo_fast_ops __read_mostly = {
  2. .id     =   "pfifo_fast",
  3. .priv_size  =   sizeof(struct pfifo_fast_priv),
  4. .enqueue    =   pfifo_fast_enqueue,
  5. .dequeue    =   pfifo_fast_dequeue,
  6. .peek       =   pfifo_fast_peek,
  7. .init       =   pfifo_fast_init,
  8. .reset      =   pfifo_fast_reset,
  9. .dump       =   pfifo_fast_dump,
  10. .owner      =   THIS_MODULE,
  11. };

在这里我们可以看到TCQ_F_CAN_BYPASS，这个flag置位，就表明数据包发送不一定非得走队列的规则，可以by pass这个规则，直接通过发送到driver，不过在一般没有阻塞的通讯状况下，有了这个flag，基本就都是直接发送出去了！

  1. static int pfifo_fast_init(struct Qdisc *qdisc, struct nlattr *opt)
  2. {
  3. int prio;
  4. struct pfifo_fast_priv *priv = qdisc_priv(qdisc);
  5. for (prio = 0; prio < PFIFO_FAST_BANDS; prio++)
  6. __skb_queue_head_init(band2list(priv, prio));
  7. /* Can by-pass the queue discipline */
  8. qdisc->flags |= TCQ_F_CAN_BYPASS;
  9. return 0;
  10. }

Enqueue

回归到正题，假设interface就是使用的pfifo_fast Qdisc规则，那么我们调用的enqueue直接走到pfifo_fast_enqueue，在里面就直接放到队列里，如果超出了最大的积攒数量就DROP掉了，返回NET_XMIT_DROP

  1. static int pfifo_fast_enqueue(struct sk_buff *skb, struct Qdisc *qdisc)
  2. {
  3. /*先是要判断一下在Q中的数据包的数量，是否超过了tx_queue_len的值*/
  4. if (skb_queue_len(&qdisc->q) < qdisc_dev(qdisc)->tx_queue_len) {
  5. int band = prio2band[skb->priority & TC_PRIO_MAX];
  6. struct pfifo_fast_priv *priv = qdisc_priv(qdisc);
  7. struct sk_buff_head *list = band2list(priv, band);
  8. priv->bitmap |= (1 << band);
  9. qdisc->q.qlen++;
  10. /*将数据包enqueue到队列中*/
  11. return __qdisc_enqueue_tail(skb, qdisc, list);
  12. }
  13. /*如果超出了tx_queue_len的最大值，说明已经积累到了最大，网络通信出现了问题，这个包被DROP*/
  14. return qdisc_drop(skb, qdisc);
  15. }

__qdisc_run

再回到__dev_xmit_skb 这个函数，enqueue完毕之后，如果这个Qdisc不是Running状态，就开启Running状态，然后调用了这个函数__qdisc_run(q);

在其中主要是调用了qdisc_restart来从队列中dequeue出封包，然后再调用sch_direct_xmit函数去直接发送封包，这个函数我们上面有分析过，就是直接发送给driver了

然后出现BUSY的就requeue到队列中。因为有可能从队列中取封包，所以这个函数可能发若干个包，要注意的是这些发送封包的过程都是出于process context

  1. void __qdisc_run(struct Qdisc *q)
  2. {
  3. int quota = weight_p;
  4. int packets;
  5. /*这个函数是调用qdisc_restart去发送Q中的数据包，packet记录这次发送了多少...*/
  6. while (qdisc_restart(q, &packets)) {
  7. /*
  8. * Ordered by possible occurrence: Postpone processing if
  9. * 1. we've exceeded packet quota
  10. * 2. another process needs the CPU;
  11. */
  12. /* 这边会有限定额度64个封包，如果超过64个就不能再次连续发了，
  13. *  需要以后执行softirq去发送了*/
  14. quota -= packets;
  15. if (quota <= 0 || need_resched()) {
  16. __netif_schedule(q);
  17. break;
  18. }
  19. }
  20. /*关闭Qdisc*/
  21. qdisc_run_end(q);
  22. }
  23. static inline int qdisc_restart(struct Qdisc *q, int *packets)
  24. {
  25. struct netdev_queue *txq;
  26. struct net_device *dev;
  27. spinlock_t *root_lock;
  28. struct sk_buff *skb;
  29. bool validate;
  30. /* Dequeue packet */
  31. /*从队列中取出skb*/
  32. skb = dequeue_skb(q, &validate, packets);
  33. if (unlikely(!skb))
  34. return 0;
  35. root_lock = qdisc_lock(q);
  36. dev = qdisc_dev(q);
  37. txq = skb_get_tx_queue(dev, skb);
  38. /*这个函数之前在将第一种状况的时候讲过，在Qdisc的状况下 也有通过这个函数直接发送的情况
  39. *实际上这个函数是在直接去发送，这个可以直接发送之前留存下来的包，所以函数的解释是
  40. *可以发送若干个封包*/
  41. return sch_direct_xmit(skb, q, dev, txq, root_lock, validate);
  42. }

ok，假设我们已经发送了若干个封包了，已经超过64个，那么会调用__netif_schedule去打开softirq利用软中断去发送在queue的封包

  1. void __netif_schedule(struct Qdisc *q)
  2. {
  3. if (!test_and_set_bit(__QDISC_STATE_SCHED, &q->state))
  4. __netif_reschedule(q);
  5. }
  6. static inline void __netif_reschedule(struct Qdisc *q)
  7. {
  8. struct softnet_data *sd;
  9. unsigned long flags;
  10. local_irq_save(flags);
  11. /*每一个CPU都有个softnet_data结构体，可以通过这个函数get到*/
  12. sd = this_cpu_ptr(&softnet_data);
  13. q->next_sched = NULL;
  14. /*将这个Q放到CPU的softnet_data的output_queue_tailp*/
  15. *sd->output_queue_tailp = q;
  16. sd->output_queue_tailp = &q->next_sched;
  17. raise_softirq_irqoff(NET_TX_SOFTIRQ);
  18. local_irq_restore(flags);
  19. }

net_tx_action

net_tx_action便是软中断的执行函数，主要是做2件事情

第一件事情就是free使用完的skb，driver一般发送完数据之后，就会调用dev_kfree_skb_irq 届时这个地方就是来free的

第二件事情就是调用qdisc_run发送数据包了，重复之前上面的分析了...

  1. static void net_tx_action(struct softirq_action *h)
  2. {
  3. struct softnet_data *sd = this_cpu_ptr(&softnet_data);
  4. /*首先判断softnet_data里的completion_queue是否为空，对于发送而言，
  5. *硬中断只是通过网卡把包发走，但是回收内存的事情是通过软中断来做的，
  6. *设备驱动发送完数据之后，会调用dev_kfree_skb_irq，不过也有的设备比较个别
  7. *自己去free，这个其实也没有什么问题的...省掉了软中断的处理*/
  8. if (sd->completion_queue) {
  9. struct sk_buff *clist;
  10. local_irq_disable();
  11. clist = sd->completion_queue;
  12. sd->completion_queue = NULL;
  13. local_irq_enable();
  14. while (clist) {
  15. struct sk_buff *skb = clist;
  16. clist = clist->next;
  17. WARN_ON(atomic_read(&skb->users));
  18. if (likely(get_kfree_skb_cb(skb)->reason == SKB_REASON_CONSUMED))
  19. trace_consume_skb(skb);
  20. else
  21. trace_kfree_skb(skb, net_tx_action);
  22. __kfree_skb(skb);
  23. }
  24. }
  25. /*这个是output_queue，主要是没有发送完成的包了，就会调用qdisc_run去发送*/
  26. if (sd->output_queue) {
  27. struct Qdisc *head;
  28. local_irq_disable();
  29. head = sd->output_queue;
  30. sd->output_queue = NULL;
  31. sd->output_queue_tailp = &sd->output_queue;
  32. local_irq_enable();
  33. while (head) {
  34. struct Qdisc *q = head;
  35. spinlock_t *root_lock;
  36. head = head->next_sched;
  37. root_lock = qdisc_lock(q);
  38. if (spin_trylock(root_lock)) {
  39. smp_mb__before_atomic();
  40. clear_bit(__QDISC_STATE_SCHED,
  41. &q->state);
  42. qdisc_run(q);
  43. spin_unlock(root_lock);
  44. } else {
  45. if (!test_bit(__QDISC_STATE_DEACTIVATED,
  46. &q->state)) {
  47. __netif_reschedule(q);
  48. } else {
  49. smp_mb__before_atomic();
  50. clear_bit(__QDISC_STATE_SCHED,
  51. &q->state);
  52. }
  53. }
  54. }
  55. }
  56. }

到这里，整个netdevice层面，Network的Tx flow已经基本分析完了，进行简单的总结！

总结

1.从宏观上来看，根本设备有无enqueue的方法，可以分成两种发送数据包的方式，第一就是有拥塞控制的数据传输，第二个就是什么都没有的直接传输到driver的,当然大部分的于外界沟通的interface都属于第一种，像loopback，tunnel一些设备就属于第二种没有enqueue的！

2. 对于有拥塞控制的数据传输，也有2条路径，第一条就是在满足3个前提条件下，直接发送数据包到硬件，和上述第二种case是一样的， 第二条就是出现拥塞的状况，就是有封包发送不成功，或者数据包量比较大的状况，这时候会用到enqueue，应该是保证顺序，所以一般q有包的状况 就都需要enqueue，然后再去dequeue发送到硬件，毕竟进程的上下文不会让你过多的占用时间，有一定的量的限制，限制条件到了就会中断发送，改用软中断的方式！

相关文章