二层（链路层）数据包发送过程分析

1790阅读　0评论2016-09-09　niao5929
分类：LINUX

二层（链路层）数据包发送过程分析

——lvyilong316

    说明：本系列博文所涉及内核版本为2.6.32
   当上层准备好一个包之后，交给链路层，链路层数据包发送主要通过dev_queue_xmit函数处理。数据包的发送可分为两种，一种是正常的传输流程，即通过网卡驱动，另一种是通过软中断（见注3）。为了理解方便，首先看一下dev_queue_xmi函数的整体调用关系图。

l dev_queue_xmit

        本函数用来将带发送的skb加入一个dev的队列（Queue），调用这个函数前必须设置好skb的device和priority，本函数可以在中断上下文中被调用。

返回值：

        返回非0（正数或负数）表示函数出错，返回0表示成功，但是并不表示数据包被成功发送出去，因为数据包可能因为限速等原因被丢掉。

        函数执行后传入的skb将被释放，所以如果想控制数据包，实现对skb的重传时需要增加skb的引用计数。

        当调用此函数时中断必须是打开的，因为BH enable必须要求IRQ enable，否则会造成死锁。

点击(此处)折叠或打开

  1. int dev_queue_xmit(struct sk_buff *skb)
  2. {
  3.     struct net_device *dev = skb->dev;
  4.     struct netdev_queue *txq;
  5.     struct Qdisc *q;
  6.     int rc = -ENOMEM;
  7.     /* GSO will handle the following emulations directly. */
  8.     if (netif_needs_gso(dev, skb))
  9.         goto gso;
  10.     if (skb_has_frags(skb) &&
  11.         !(dev->features & NETIF_F_FRAGLIST) &&
  12.         __skb_linearize(skb))
  13.         goto out_kfree_skb;
  14. //如果skb有分片但是发送设备不支持分片，或分片中有分片在高端内存但发送设备不支持DMA，需要将所有段重新组合成一个段 ，这里__skb_linearize其实就是__pskb_pull_tail(skb, skb->data_len),这个函数基本上等同于pskb_may_pull ，pskb_may_pull的作用就是检测skb对应的主buf中是否有足够的空间来pull出len长度，如果不够就重新分配skb并将frags中的数据拷贝入新分配的主buff中，而这里将参数len设置为skb->datalen， 也就是会将所有的数据全部拷贝到主buff中，以这种方式完成skb的线性化。
  15.     if (skb_shinfo(skb)->nr_frags &&
  16.         (!(dev->features & NETIF_F_SG) || illegal_highdma(dev, skb)) &&
  17.         __skb_linearize(skb))
  18.         goto out_kfree_skb;
  19. //如果数据包没有被计算校验和并且发送设备不支持这个协议的校验，则在此进行校验和的计算（注1）。如果上面已经线性化了一次，这里的__skb_linearize就会直接返回,注意区别frags和frag_list，前者是将多的数据放到单独分配的页面中，sk_buff只有一个。而后者则是连接多个sk_buff
  20.     if (skb->ip_summed == CHECKSUM_PARTIAL) {
  21.         skb_set_transport_header(skb, skb->csum_start -
  22.                           skb_headroom(skb));
  23.         if (!dev_can_checksum(dev, skb) && skb_checksum_help(skb))
  24.             goto out_kfree_skb;
  25.     }
  26. gso:
  27. //关闭软中断,禁止cpu抢占
  28.     rcu_read_lock_bh();
  29. //选择一个发送队列，如果设备提供了select_queue回调函数就使用它，否则由内核选择一个队列,这里只是Linux内核多队列的实现，但是要真正的使用都队列，需要网卡支持多队列才可以，一般的网卡都只有一个队列。在调用alloc_etherdev分配net_device是，设置队列的个数
  30. txq = dev_pick_tx(dev, skb);
  31. // 从netdev_queue结构上获取设备的qdisc
  32.     q = rcu_dereference(txq->qdisc);
  33. //如果该设备有队列可用，就调用__dev_xmit_skb
  34.     if (q->enqueue) {
  35.         rc = __dev_xmit_skb(skb, q, dev, txq);
  36.         goto out;
  37.     }
  38. //下面的处理是在没有发送队列的情况,软设备一般没有发送队列：如lo、tunnle；我们所要做的就是直接调用驱动的hard_start_xmit将它发送出去 如果发送失败就直接丢弃，因为没有队列可以保存它
  39.     if (dev->flags & IFF_UP) { //确定设备是否开启
  40.         int cpu = smp_processor_id(); /* ok because BHs are off */
  41.         if (txq->xmit_lock_owner != cpu) {//是否在同一个cpu上
  42.             HARD_TX_LOCK(dev, txq, cpu);
  43.             if (!netif_tx_queue_stopped(txq)) {//确定队列是运行状态
  44.                 rc = NET_XMIT_SUCCESS;
  45.                 if (!dev_hard_start_xmit(skb, dev, txq)) {
  46.                     HARD_TX_UNLOCK(dev, txq);
  47.                     goto out;
  48.                 }
  49.             }
  50.             HARD_TX_UNLOCK(dev, txq);
  51.             if (net_ratelimit())
  52.                 printk(KERN_CRIT "Virtual device %s asks to "
  53.                        "queue packet!\n", dev->name);
  54.         } else {// txq->xmit_lock_owner == cpu的情况，说明发生递归
  55.             if (net_ratelimit())
  56.                 printk(KERN_CRIT "Dead loop on virtual device "
  57.                        "%s, fix it urgently!\n", dev->name);
  58.         }
  59.     }
  60.     rc = -ENETDOWN;
  61.     rcu_read_unlock_bh();
  62. out_kfree_skb:
  63.     kfree_skb(skb);
  64.     return rc;
  65. out:
  66.     rcu_read_unlock_bh();
  67.     return rc;
  68. }

l __dev_xmit_skb

       __dev_xmit_skb函数主要做两件事情：

（1）如果流控对象为空的，试图直接发送数据包。

（2）如果流控对象不空，将数据包加入流控对象，并运行流控对象。

点击(此处)折叠或打开

  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);//见注2
  6.     int rc;
  7.     spin_lock(root_lock); //锁qdisc
  8.     if (unlikely(test_bit(__QDISC_STATE_DEACTIVATED, &q->state))) {//判断队列是否失效
  9.         kfree_skb(skb);
  10.         rc = NET_XMIT_DROP;
  11.     } else if ((q->flags & TCQ_F_CAN_BYPASS) && !qdisc_qlen(q) &&
  12.            !test_and_set_bit(__QDISC_STATE_RUNNING, &q->state)) {
  13.         /*
  14.          * This is a work-conserving queue; there are no old skbs
  15.          * waiting to be sent out; and the qdisc is not running -
  16.          * xmit the skb directly.
  17.          */
  18.         __qdisc_update_bstats(q, skb->len);
  19.         if (sch_direct_xmit(skb, q, dev, txq, root_lock))
  20.             __qdisc_run(q);
  21.         else
  22.             clear_bit(__QDISC_STATE_RUNNING, &q->state);
  23.         rc = NET_XMIT_SUCCESS;
  24.     } else {
  25.         rc = qdisc_enqueue_root(skb, q);
  26.         qdisc_run(q);
  27.     }
  28.     spin_unlock(root_lock);
  29.     return rc;
  30. }

l qdisc_run

有两个时机将会调用qdisc_run()：

1．__dev_xmit_skb()

2. 软中断服务线程NET_TX_SOFTIRQ

点击(此处)折叠或打开

  1. static inline void qdisc_run(struct Qdisc *q)
  2. {
  3.     if (!test_and_set_bit(__QDISC_STATE_RUNNING, &q->state))//将队列设置为运行状态
  4.         __qdisc_run(q);
  5. }

l __qdisc_run

点击(此处)折叠或打开

  1. void __qdisc_run(struct Qdisc *q)
  2. {
  3.     unsigned long start_time = jiffies;
  4.     while (qdisc_restart(q)) { //返回值大于0，说明流控对象非空
  5.         /*如果发现本队列运行的时间太长了，将会停止队列的运行，并将队列加入output_queue链表头
  6.          * Postpone processing if (延迟处理)
  7.          * 1. another process needs the CPU;
  8.          * 2. we've been doing it for too long.
  9.          */
  10.         if (need_resched() || jiffies != start_time) { //已经不允许继续运行本流控对象
  11.             __netif_schedule(q); //将本qdisc加入每cpu变量softnet_data的output_queue链表中
  12.             break;
  13.         }
  14.     }
  15.   //清除队列的运行标识
  16.     clear_bit(__QDISC_STATE_RUNNING, &q->state);
  17. }

循环调用qdisc_restart发送数据,下面这个函数qdisc_restart是真正发送数据包的函数,它从队列上取下一个帧，然后尝试将它发送出去,若发送失败则一般是重新入队。

此函数返回值为：发送成功时返回剩余队列长度,发送失败时返回0（若发送成功且剩余队列长度为0也返回0）

l qdisc_restart

__QDISC_STATE_RUNNING状态保证同一时刻只有一个cpu在处理这个qdisc，qdisc_lock(q)用来保证对这个队列的顺序访问。

通常netif_tx_lock用来确保本设备驱动的顺序（独占）访问的，qdisc_lock(q)用来保证qdisc的顺序访问，这两个是互斥的，获得其中一个必须释放另一个。

点击(此处)折叠或打开

  1. static inline int qdisc_restart(struct Qdisc *q)
  2. {
  3.     struct netdev_queue *txq;
  4.     struct net_device *dev;
  5.     spinlock_t *root_lock;
  6.     struct sk_buff *skb;
  7.     /* Dequeue packet */
  8.     skb = dequeue_skb(q); //一开始就调用dequeue函数
  9.     if (unlikely(!skb))
  10.         return 0; //返回0说明队列是空的或者被限制
  11.     root_lock = qdisc_lock(q);
  12.     dev = qdisc_dev(q);
  13.     txq = netdev_get_tx_queue(dev, skb_get_queue_mapping(skb));
  14.     return sch_direct_xmit(skb, q, dev, txq, root_lock); //用于发送数据包
  15. }

l sch_direct_xmit

       发送一个skb，将队列置为__QDISC_STATE_RUNNING状态，保证只有一个cpu运行这个函数，返回0表示队列为空或者发送受限，大于0表示队列非空。

点击(此处)折叠或打开

  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)
  4. {
  5.     int ret = NETDEV_TX_BUSY;
  6.     spin_unlock(root_lock);// release qdisc,因为后面要获取设备锁
  7.    // 调用__netif_tx_lockà spin_lock(&txq->_xmit_lock,，保证设备驱动的独占访问
  8.     HARD_TX_LOCK(dev, txq, smp_processor_id());
  9.     if (!netif_tx_queue_stopped(txq) && //设备没有被停止，且发送队列没有被冻结
  10.         !netif_tx_queue_frozen(txq))
  11.         ret = dev_hard_start_xmit(skb, dev, txq); //发送数据包
  12.     HARD_TX_UNLOCK(dev, txq); // 调用__netif_tx_unlock
  13.     spin_lock(root_lock);
  14.     switch (ret) {
  15.     case NETDEV_TX_OK: //如果设备成功将数据包发送出去
  16.         ret = qdisc_qlen(q); //返回剩余的队列长度
  17.         break;
  18.     case NETDEV_TX_LOCKED: //获取设备锁失败
  19.         ret = handle_dev_cpu_collision(skb, txq, q);
  20.         break;
  21.     default: //设备繁忙，重新入队发送（利用softirq）
  22.         if (unlikely (ret != NETDEV_TX_BUSY && net_ratelimit()))
  23.             printk(KERN_WARNING "BUG %s code %d qlen %d\n",
  24.                    dev->name, ret, q->q.qlen);
  25.         ret = dev_requeue_skb(skb, q);
  26.         break;
  27.     }
  28.     if (ret && (netif_tx_queue_stopped(txq) ||
  29.             netif_tx_queue_frozen(txq)))
  30.         ret = 0;
  31.     return ret;
  32. }

l dev_hard_start_xmit

点击(此处)折叠或打开

  1. int dev_hard_start_xmit(struct sk_buff *skb, struct net_device *dev,
  2.             struct netdev_queue *txq)
  3. {
  4.     const struct net_device_ops *ops = dev->netdev_ops;
  5.     int rc;
  6. if (likely(!skb->next)) {
  7. //从这里可以看出，对于每一个发送的包也会发给ptype_all一份， 而packet套接字创建时对于proto为ETH_P_ALL的会在ptype_all中注册一个成员,因此对于协议号为ETH_P_ALL的packet套接字来说，发送和接受的数据都能收到
  8.         if (!list_empty(&ptype_all))
  9.             dev_queue_xmit_nit(skb, dev);
  10.         if (netif_needs_gso(dev, skb)) {
  11.             if (unlikely(dev_gso_segment(skb)))
  12.                 goto out_kfree_skb;
  13.             if (skb->next)
  14.                 goto gso;
  15.         }
  16.        //如果发送设备不需要skb->dst，则在此将其释放
  17.         if (dev->priv_flags & IFF_XMIT_DST_RELEASE)
  18.             skb_dst_drop(skb);
  19.        //调用设备注册的发送函数，即dev->netdev_ops-> ndo_start_xmit(skb, dev)
  20.         rc = ops->ndo_start_xmit(skb, dev);
  21.         if (rc == NETDEV_TX_OK)
  22. txq_trans_update(txq);
  23.         return rc;
  24.     }
  25. gso:
  26. ……
  27. }

l dev_queue_xmit_nit

点击(此处)折叠或打开

  1. static void dev_queue_xmit_nit(struct sk_buff *skb, struct net_device *dev)
  2. {
  3.     struct packet_type *ptype;
  4. #ifdef CONFIG_NET_CLS_ACT
  5.     if (!(skb->tstamp.tv64 && (G_TC_FROM(skb->tc_verd) & AT_INGRESS)))
  6.         net_timestamp(skb); //记录该数据包输入的时间戳
  7. #else
  8.     net_timestamp(skb);
  9. #endif
  10.     rcu_read_lock();
  11.     list_for_each_entry_rcu(ptype, &ptype_all, list) {
  12.         /* Never send packets back to the socket they originated from */
  13.        //遍历ptype_all链表，查找所有符合输入条件的原始套接口，并循环将数据包输入到满足条件的套接口
  14.         if ((ptype->dev == dev || !ptype->dev) &&
  15.             (ptype->af_packet_priv == NULL ||
  16. (struct sock *)ptype->af_packet_priv != skb->sk)) {
  17.        //由于该数据包是额外输入到这个原始套接口的，因此需要克隆一个数据包
  18.             struct sk_buff *skb2 = skb_clone(skb, GFP_ATOMIC);
  19.             if (!skb2)
  20.                 break;
  21.             /* skb->nh should be correctly(确保头部偏移正确)
  22.                set by sender, so that the second statement is
  23.                just protection against buggy protocols.
  24.              */
  25.             skb_reset_mac_header(skb2);
  26.             if (skb_network_header(skb2) < skb2->data ||
  27.                 skb2->network_header > skb2->tail) {
  28.                 if (net_ratelimit())//net_ratelimit用来保证网络代码中printk的频率
  29.                     printk(KERN_CRIT "protocol %04x is "
  30.                            "buggy, dev %s\n",
  31.                            skb2->protocol, dev->name);
  32.                 skb_reset_network_header(skb2); //重新设置L3头部偏移
  33.             }
  34.             skb2->transport_header = skb2->network_header;
  35.             skb2->pkt_type = PACKET_OUTGOING;
  36.             ptype->func(skb2, skb->dev, ptype, skb->dev);//调用协议(ptype_all)接受函数
  37.         }
  38.     }
  39.     rcu_read_unlock();
  40. }

? 环回设备

    对于环回设备loopback，设备的ops->ndo_start_xmit被初始化为loopback_xmit函数。

点击(此处)折叠或打开

  1. static const struct net_device_ops loopback_ops = {
  2.     .ndo_init = loopback_dev_init,
  3.     .ndo_start_xmit= loopback_xmit,
  4.     .ndo_get_stats = loopback_get_stats,
  5. };

drivers/net/loopback.c

点击(此处)折叠或打开

  1. static netdev_tx_t loopback_xmit(struct sk_buff *skb,
  2.                  struct net_device *dev)
  3. {
  4.     struct pcpu_lstats *pcpu_lstats, *lb_stats;
  5.     int len;
  6.     skb_orphan(skb);
  7.     skb->protocol = eth_type_trans(skb, dev);
  8.     /* it's OK to use per_cpu_ptr() because BHs are off */
  9.     pcpu_lstats = dev->ml_priv;
  10.     lb_stats = per_cpu_ptr(pcpu_lstats, smp_processor_id());
  11.     len = skb->len;
  12.     if (likely(netif_rx(skb) == NET_RX_SUCCESS)) { //直接调用了netif_rx进行了接收处理
  13.         lb_stats->bytes += len;
  14.         lb_stats->packets++;
  15.     } else
  16.         lb_stats->drops++;
  17.     return NETDEV_TX_OK;
  18. }


1. CHECKSUM_PARTIAL表示使用硬件checksum ,L4层的伪头的校验已经完毕,并且已经加入uh->check字段中,此时只需要设备计算整个头4层头的校验值。


2. 整个数据包发送逻辑中会涉及到三个用于互斥访问的代码：

（1）spinlock_t *root_lock = qdisc_lock(q);

（2）test_and_set_bit(__QDISC_STATE_RUNNING, &q->state)

（3）__netif_tx_lockà spin_lock(&txq->_xmit_lock)

     其中（1）（3）分别对应一个spinlock，（2）对应一个队列状态。在了解代码中如何使用这三个同步方法时，首先看一下相关数据结构的关系，如下。

    图中绿色部分表示（1）（3）两处spinlock。首先看（1）处对应的代码：

点击(此处)折叠或打开

  1. static inline spinlock_t *qdisc_lock(struct Qdisc *qdisc)
  2. {
  3.     return &qdisc->q.lock;
  4. }

所以root_lock是用于控制qdisc中skb队列访问的锁，当需要对skb队列进行enqueue、dequeue、requeue时，就需要加锁。

__QDISC_STATE_RUNNING标志用于保证一个流控对象（qdisc）不会同时被多个cpu访问。

而（3）处的spinlock，即struct netdev_queue中的_xmit_lock，则用于保证dev的注册函数的互斥访问，即deriver的同步。

另外，内核代码注释中写到，（1）和（3）是互斥的，获得（1）处的锁时必须先保证释放（3）处的锁，反之亦然，为什么要这样还没有想明白。。。。哪位大神知道还望指点

3. 已经有了dev_queue_xmit函数，为什么还需要软中断来发送呢？

我们可以看到在dev_queue_xmit中将skb进行了一些处理(比如合并成一个包，计算校验和等),处理完的skb是可以直接发送的了，这时dev_queue_xmit也会先将skb入队（skb一般都是在这个函数中入队的）,并且调用qdisc_run尝试发送，但是有可能发送失败，这时就将skb重新入队，调度软中断，并且自己直接返回。

软中断只是发送队列中的skb以及释放已经发送的skb,它无需再对skb进行线性化或者校验和处理。另外在队列被停止的情况下，dev_queue_xmit仍然可以把包加入队列，但是不能发送，这样在队列被唤醒的时候就需要通过软中断来发送停止期间积压的包。简而言之，dev_queue_xmit是对skb做些最后的处理并且第一次尝试发送,软中断是将前者发送失败或者没发完的包发送出去。（其实发送软中断还有一个作用，就是释放已经发送的包，因为某些情况下发送是在硬件中断中完成的，为了提高硬件中断处理效率，内核提供一种方式将释放skb放到软中断中进行，这时只要调用dev_kfree_skb_irq，它将skb加入softnet_data的completion_queue中，然后开启发送软中断，net_tx_action会在软中断中将completion_queue中的skb全部释放掉）

上一篇：ceph最新版本JEWEL（宝石）：只支持debian8，7以下的已经不行了
下一篇：使用linux heartbeat 的一些经验