注: この一連のブログ投稿に含まれるカーネル バージョンは 2.6.32 です。
上位層がパケットを準備した後、リンク層に引き渡されるとき、リンク層のデータパケット送信は主に dev_queue_xmit 関数を通じて処理されます。データ パケットの送信は 2 つのタイプに分類できます。1 つは通常の送信プロセス、つまりネットワーク カード ドライバーを介した送信プロセスで、もう 1 つはソフト割り込みを介した送信プロセスです (注 3 を参照)。理解を容易にするために、まず dev_queue_xmi 関数の全体的な呼び出し図を見てください。
ldev_queue_xmit
この関数は、送信された skb を開発キュー (キュー) に追加するために使用されます。この関数を呼び出す前に、skb のデバイスと優先順位を設定する必要があります。
戻り値:
0 以外の値 (正または負の数値) が返された場合は、関数にエラーがあることを示します。0 が返された場合は成功を示しますが、データ パケットが失われる可能性があるため、データ パケットが正常に送信されたことを意味しません。速度制限やその他の理由により。
受信したskbは関数実行後に解放されるため、データパケットを制御したい場合はskb再送時にskbの参照カウントを増やす必要があります。
BHenable は IRQenable を必要とするため、この関数を呼び出すときは割り込みをオンにする必要があります。オンにしないとデッドロックが発生します。
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>int dev_queue_xmit(struct sk_buff *skb)<br /> </li><li>{<br /></li><li>struct net_device *dev = skb->dev;<br /></li><li>struct netdev_queue *txq;<br /></li><li>struct Qdisc *q;<br /></li><li>int rc = -ENOMEM;<br /></li><li>/* GSO will handle the following emulations directly. */<br /></li><li>if (netif_needs_gso(dev, skb))<br /></li><li>goto gso;<br /></li><li>if (skb_has_frags(skb) &&<br /></li><li>!(dev->features & NETIF_F_FRAGLIST) &&<br /></li><li>__skb_linearize(skb))<br /></li><li>goto out_kfree_skb;<br /></li><li>//如果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的线性化。<br /></li><li>if (skb_shinfo(skb)->nr_frags &&<br /></li><li>(!(dev->features & NETIF_F_SG) || illegal_highdma(dev, skb)) &&<br /></li><li>__skb_linearize(skb))<br /></li><li>goto out_kfree_skb;<br /></li><li>//如果数据包没有被计算校验和并且发送设备不支持这个协议的校验，则在此进行校验和的计算（注1）。如果上面已经线性化了一次，这里的__skb_linearize就会直接返回,注意区别frags和frag_list，前者是将多的数据放到单独分配的页面中，sk_buff只有一个。而后者则是连接多个sk_buff <br /></li><li>if (skb->ip_summed == CHECKSUM_PARTIAL) {<br /></li><li>skb_set_transport_header(skb, skb->csum_start -<br /></li><li>skb_headroom(skb));<br /></li><li>if (!dev_can_checksum(dev, skb) && skb_checksum_help(skb))<br /></li><li>goto out_kfree_skb;<br /></li><li>}<br /></li><li>gso:<br /></li><li>//关闭软中断,禁止cpu抢占<br /></li><li>rcu_read_lock_bh();<br /></li><li>//选择一个发送队列，如果设备提供了select_queue回调函数就使用它，否则由内核选择一个队列,这里只是Linux内核多队列的实现，但是要真正的使用都队列，需要网卡支持多队列才可以，一般的网卡都只有一个队列。在调用alloc_etherdev分配net_device是，设置队列的个数<br /></li><li>txq = dev_pick_tx(dev, skb);<br /></li><li>// 从netdev_queue结构上获取设备的qdisc <br /></li><li>q = rcu_dereference(txq->qdisc);<br /></li><li>//如果该设备有队列可用，就调用__dev_xmit_skb <br /></li><li>if (q->enqueue) {<br /></li><li>rc = __dev_xmit_skb(skb, q, dev, txq);<br /></li><li>goto out;<br /></li><li>}<br /></li><li>//下面的处理是在没有发送队列的情况,软设备一般没有发送队列：如lo、tunnle；我们所要做的就是直接调用驱动的hard_start_xmit将它发送出去 如果发送失败就直接丢弃，因为没有队列可以保存它 <br /></li><li>if (dev->flags & IFF_UP) { //确定设备是否开启<br /></li><li>int cpu = smp_processor_id(); /* ok because BHs are off */<br /></li><li>if (txq->xmit_lock_owner != cpu) {//是否在同一个cpu上<br /></li><li>HARD_TX_LOCK(dev, txq, cpu);<br /></li><li>if (!netif_tx_queue_stopped(txq)) {//确定队列是运行状态<br /></li><li>rc = NET_XMIT_SUCCESS;<br /></li><li>if (!dev_hard_start_xmit(skb, dev, txq)) {<br /></li><li>HARD_TX_UNLOCK(dev, txq);<br /></li><li>goto out;<br /></li><li>}<br /></li><li>}<br /></li><li>HARD_TX_UNLOCK(dev, txq);<br /></li><li>if (net_ratelimit())<br /></li><li>printk(KERN_CRIT "Virtual device %s asks to "<br /></li><li>"queue packet!\n", dev->name);<br /></li><li>} else {// txq->xmit_lock_owner == cpu的情况，说明发生递归<br /></li><li>if (net_ratelimit())<br /></li><li>printk(KERN_CRIT "Dead loop on virtual device "<br /></li><li>"%s, fix it urgently!\n", dev->name);<br /></li><li>}<br /></li><li>}<br /></li><li>rc = -ENETDOWN;<br /></li><li>rcu_read_unlock_bh();<br /></li><li>out_kfree_skb:<br /></li><li>kfree_skb(skb);<br /></li><li>return rc;<br /></li><li>out:<br /></li><li>rcu_read_unlock_bh();<br /></li><li>return rc;<br /></li><li>}</li></ol>
l__dev_xmit_skb
__dev_xmit_skb 関数は主に 2 つのことを行います:
(1) フロー制御オブジェクトが空の場合は、データ パケットを直接送信しようとします。
(2) フロー制御オブジェクトが空でない場合は、データ パケットをフロー制御オブジェクトに追加し、フロー制御オブジェクトを実行します。
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>static inline int __dev_xmit_skb(struct sk_buff *skb, struct Qdisc *q,<br /> </li><li>struct net_device *dev,<br /></li><li>struct netdev_queue *txq)<br /></li><li>{<br /></li><li>spinlock_t *root_lock = qdisc_lock(q);//见注2<br /></li><li>int rc;<br /></li><li>spin_lock(root_lock); //锁qdisc<br /></li><li>if (unlikely(test_bit(__QDISC_STATE_DEACTIVATED, &q->state))) {//判断队列是否失效<br /></li><li>kfree_skb(skb);<br /></li><li>rc = NET_XMIT_DROP;<br /></li><li>} else if ((q->flags & TCQ_F_CAN_BYPASS) && !qdisc_qlen(q) &&<br /></li><li>!test_and_set_bit(__QDISC_STATE_RUNNING, &q->state)) {<br /></li><li>/*<br /></li><li>* This is a work-conserving queue; there are no old skbs<br /></li><li>* waiting to be sent out; and the qdisc is not running -<br /></li><li>* xmit the skb directly.<br /></li><li>*/<br /></li><li>__qdisc_update_bstats(q, skb->len);<br /></li><li>if (sch_direct_xmit(skb, q, dev, txq, root_lock))<br /></li><li>__qdisc_run(q);<br /></li><li>else<br /></li><li>clear_bit(__QDISC_STATE_RUNNING, &q->state);<br /></li><li>rc = NET_XMIT_SUCCESS;<br /></li><li>} else {<br /></li><li>rc = qdisc_enqueue_root(skb, q);<br /></li><li>qdisc_run(q);<br /></li><li>}<br /></li><li>spin_unlock(root_lock);<br /></li><li>return rc;<br /></li><li>}</li></ol>
lqdisc_run
qdisc_run() を呼び出す機会は 2 つあります:
1． __dev_xmit_skb()
2. ソフト割り込みサービス スレッド NET_TX_SOFTIRQ
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>static inline void qdisc_run(struct Qdisc *q)<br /> </li><li>{<br /></li><li>if (!test_and_set_bit(__QDISC_STATE_RUNNING, &q->state))//将队列设置为运行状态<br /></li><li>__qdisc_run(q);<br /></li><li>}</li></ol>l__qdisc_run
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>void __qdisc_run(struct Qdisc *q)<br /> </li><li>{<br /></li><li>unsigned long start_time = jiffies;<br /></li><li>while (qdisc_restart(q)) { //返回值大于0，说明流控对象非空<br /></li><li>/*如果发现本队列运行的时间太长了，将会停止队列的运行，并将队列加入output_queue链表头<br /></li><li>* Postpone processing if (延迟处理)<br /></li><li>* 1. another process needs the CPU;<br /></li><li>* 2. we've been doing it for too long.<br /></li><li>*/<br /></li><li>if (need_resched() || jiffies != start_time) { //已经不允许继续运行本流控对象<br /></li><li>__netif_schedule(q); //将本qdisc加入每cpu变量softnet_data的output_queue链表中<br /></li><li>break;<br /></li><li>}<br /></li><li>}<br /></li><li>//清除队列的运行标识<br /></li><li>clear_bit(__QDISC_STATE_RUNNING, &q->state);<br /></li><li>}</li></ol>は、データを送信するために qdisc_restart を呼び出す関数です。これは、キューからフレームを取得し、送信が失敗すると、通常はキューに再度入ります。
この関数の戻り値は、送信成功の場合は残りのキュー長を返し、送信失敗の場合は0を返します（送信成功で残りのキュー長が0の場合も0を返します）
lqdisc_restart
__QDISC_STATE_RUNNING 状態は、同時に CPU が 1 つだけであることを保証します。この qdisc を処理するとき、このキューへの順次アクセスを保証するために qdisc_lock(q) が使用されます。
通常、netif_tx_lock はこのデバイスドライバーへの順次 (排他的) アクセスを保証するために使用され、qdisc_lock(q) は qdisc への順次アクセスを保証するために使用されます。一方を取得するには、他方を解放する必要があります。
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>static inline int qdisc_restart(struct Qdisc *q)<br /> </li><li>{<br /></li><li>struct netdev_queue *txq;<br /></li><li>struct net_device *dev;<br /></li><li>spinlock_t *root_lock;<br /></li><li>struct sk_buff *skb;<br /></li><li>/* Dequeue packet */<br /></li><li>skb = dequeue_skb(q); //一开始就调用dequeue函数<br /></li><li>if (unlikely(!skb))<br /></li><li>return 0; //返回0说明队列是空的或者被限制<br /></li><li>root_lock = qdisc_lock(q);<br /></li><li>dev = qdisc_dev(q);<br /></li><li>txq = netdev_get_tx_queue(dev, skb_get_queue_mapping(skb));<br /></li><li>return sch_direct_xmit(skb, q, dev, txq, root_lock); //用于发送数据包<br /></li><li>}</li></ol>lsch_direct_xmit
skb を送信し、キューを __QDISC_STATE_RUNNING 状態に設定して、1 つの CPU のみがこの関数を実行するようにします。0 を返す場合は、キューが空であるか送信が制限されていることを意味し、0 より大きい場合はキューが空ではないことを意味します。 。
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>int sch_direct_xmit(struct sk_buff *skb, struct Qdisc *q,<br /> </li><li>struct net_device *dev, struct netdev_queue *txq,<br /></li><li>spinlock_t *root_lock)<br /></li><li>{<br /></li><li>int ret = NETDEV_TX_BUSY;<br /></li><li>spin_unlock(root_lock);// release qdisc,因为后面要获取设备锁<br /></li><li>// 调用__netif_tx_lockà spin_lock(&txq->_xmit_lock,，保证设备驱动的独占访问<br /></li><li>HARD_TX_LOCK(dev, txq, smp_processor_id());<br /></li><li>if (!netif_tx_queue_stopped(txq) && //设备没有被停止，且发送队列没有被冻结<br /></li><li>!netif_tx_queue_frozen(txq))<br /></li><li>ret = dev_hard_start_xmit(skb, dev, txq); //发送数据包<br /></li><li>HARD_TX_UNLOCK(dev, txq); // 调用__netif_tx_unlock<br /></li><li>spin_lock(root_lock);<br /></li><li>switch (ret) {<br /></li><li>case NETDEV_TX_OK: //如果设备成功将数据包发送出去<br /></li><li>ret = qdisc_qlen(q); //返回剩余的队列长度<br /></li><li>break;<br /></li><li>case NETDEV_TX_LOCKED: //获取设备锁失败<br /></li><li>ret = handle_dev_cpu_collision(skb, txq, q);<br /></li><li>break;<br /></li><li>default: //设备繁忙，重新入队发送（利用softirq）<br /></li><li>if (unlikely (ret != NETDEV_TX_BUSY && net_ratelimit()))<br /></li><li>printk(KERN_WARNING "BUG %s code %d qlen %d\n",<br /></li><li>dev->name, ret, q->q.qlen);<br /></li><li>ret = dev_requeue_skb(skb, q);<br /></li><li>break;<br /></li><li>}<br /></li><li>if (ret && (netif_tx_queue_stopped(txq) ||<br /></li><li>netif_tx_queue_frozen(txq)))<br /></li><li>ret = 0;<br /></li><li>return ret;<br /></li><li>}</li></ol>
ldev_hard_start_xmit
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>int dev_hard_start_xmit(struct sk_buff *skb, struct net_device *dev,<br /> </li><li>struct netdev_queue *txq)<br /></li><li>{<br /></li><li>const struct net_device_ops *ops = dev->netdev_ops;<br /></li><li>int rc;<br /></li><li>if (likely(!skb->next)) {<br /></li><li>//从这里可以看出，对于每一个发送的包也会发给ptype_all一份， 而packet套接字创建时对于proto为ETH_P_ALL的会在ptype_all中注册一个成员,因此对于协议号为ETH_P_ALL的packet套接字来说，发送和接受的数据都能收到<br /></li><li>if (!list_empty(&ptype_all))<br /></li><li>dev_queue_xmit_nit(skb, dev);<br /></li><li>if (netif_needs_gso(dev, skb)) {<br /></li><li>if (unlikely(dev_gso_segment(skb)))<br /></li><li>goto out_kfree_skb;<br /></li><li>if (skb->next)<br /></li><li>goto gso;<br /></li><li>}<br /></li><li>//如果发送设备不需要skb->dst，则在此将其释放<br /></li><li>if (dev->priv_flags & IFF_XMIT_DST_RELEASE)<br /></li><li>skb_dst_drop(skb);<br /></li><li>//调用设备注册的发送函数，即dev->netdev_ops-> ndo_start_xmit(skb, dev)<br /></li><li>rc = ops->ndo_start_xmit(skb, dev);<br /></li><li>if (rc == NETDEV_TX_OK)<br /></li><li>txq_trans_update(txq);<br /></li><li>return rc;<br /></li><li>}<br /></li><li>gso:<br /></li><li>……<br /></li><li>}</li></ol>ldev_queue_xmit_nit
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>static void dev_queue_xmit_nit(struct sk_buff *skb, struct net_device *dev)<br /> </li><li>{<br /></li><li>struct packet_type *ptype;<br /></li><li>#ifdef CONFIG_NET_CLS_ACT<br /></li><li>if (!(skb->tstamp.tv64 && (G_TC_FROM(skb->tc_verd) & AT_INGRESS)))<br /></li><li>net_timestamp(skb); //记录该数据包输入的时间戳<br /></li><li>#else<br /></li><li>net_timestamp(skb);<br /></li><li>#endif<br /></li><li>rcu_read_lock();<br /></li><li>list_for_each_entry_rcu(ptype, &ptype_all, list) {<br /></li><li>/* Never send packets back to the socket they originated from */<br /></li><li>//遍历ptype_all链表，查找所有符合输入条件的原始套接口，并循环将数据包输入到满足条件的套接口<br /></li><li>if ((ptype->dev == dev || !ptype->dev) &&<br /></li><li>(ptype->af_packet_priv == NULL ||<br /></li><li>(struct sock *)ptype->af_packet_priv != skb->sk)) {<br /></li><li>//由于该数据包是额外输入到这个原始套接口的，因此需要克隆一个数据包<br /></li><li>struct sk_buff *skb2 = skb_clone(skb, GFP_ATOMIC);<br /></li><li>if (!skb2)<br /></li><li>break;<br /></li><li>/* skb->nh should be correctly(确保头部偏移正确)<br /></li><li>set by sender, so that the second statement is<br /></li><li>just protection against buggy protocols.<br /></li><li>*/<br /></li><li>skb_reset_mac_header(skb2);<br /></li><li>if (skb_network_header(skb2) < skb2->data ||<br /></li><li>skb2->network_header > skb2->tail) {<br /></li><li>if (net_ratelimit())//net_ratelimit用来保证网络代码中printk的频率<br /></li><li>printk(KERN_CRIT "protocol %04x is "<br /></li><li>"buggy, dev %s\n",<br /></li><li>skb2->protocol, dev->name);<br /></li><li>skb_reset_network_header(skb2); //重新设置L3头部偏移<br /></li><li>}<br /></li><li>skb2->transport_header = skb2->network_header;<br /></li><li>skb2->pkt_type = PACKET_OUTGOING;<br /></li><li>ptype->func(skb2, skb->dev, ptype, skb->dev);//调用协议(ptype_all)接受函数<br /></li><li>}<br /></li><li>}<br /></li><li>rcu_read_unlock();<br /></li><li>}</li></ol>
?ループバックデバイス
ループバックデバイスのループバックの場合、デバイスの ops->ndo_start_xmit は、loopback_xmit 関数に初期化されます。
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>static const struct net_device_ops loopback_ops = {<br /> </li><li>.ndo_init = loopback_dev_init,<br /></li><li>.ndo_start_xmit= loopback_xmit,<br /></li><li>.ndo_get_stats = loopback_get_stats,<br /></li><li>};</li></ol>drivers/net/loopback.c
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>static netdev_tx_t loopback_xmit(struct sk_buff *skb,<br /> </li><li>struct net_device *dev)<br /></li><li>{<br /></li><li>struct pcpu_lstats *pcpu_lstats, *lb_stats;<br /></li><li>int len;<br /></li><li>skb_orphan(skb);<br /></li><li>skb->protocol = eth_type_trans(skb, dev);<br /></li><li>/* it's OK to use per_cpu_ptr() because BHs are off */<br /></li><li>pcpu_lstats = dev->ml_priv;<br /></li><li>lb_stats = per_cpu_ptr(pcpu_lstats, smp_processor_id());<br /></li><li>len = skb->len;<br /></li><li>if (likely(netif_rx(skb) == NET_RX_SUCCESS)) { //直接调用了netif_rx进行了接收处理<br /></li><li>lb_stats->bytes += len;<br /></li><li>lb_stats->packets++;<br /></li><li>} else<br /></li><li>lb_stats->drops++;<br /></li><li>return NETDEV_TX_OK;<br /></li><li>}</li></ol>
2. データ パケット送信ロジック全体には、相互排他的アクセスのための 3 つのコードが含まれます:
(1) stop_and_set_bit(__QDISC_STATE_RUNNING,&q- >state)
(3)__netif_tx_lockàspin_lock(&txq->_xmit_lock)
ここで、(1) (3) はそれぞれスピンロックに対応し、(2) はキューの状態に対応します。コード内でこれら 3 つの同期メソッドを使用する方法を理解するときは、まず、次のように、関連するデータ構造間の関係に注目してください。
写真の緑色の部分は 2 つのスピンロック (1) (3) を表します。まず、(1) に対応するコードを見てください:
<ol style="margin:0 1px 0 0px;padding-left:40px;" start="1" class="dp-css"><li>static inline spinlock_t *qdisc_lock(struct Qdisc *qdisc)<br /> </li><li>{<br /></li><li>return &qdisc->q.lock;<br /></li><li>}</li></ol>つまり、root_lock は、qdisc の skb キューへのアクセスを制御するために使用されるロックであり、skb キューをエンキュー、デキュー、および再キューする必要がある場合、ロックする必要があります。
__QDISC_STATE_RUNNING フラグは、フロー制御オブジェクト (qdisc) が複数の CPU によって同時にアクセスされないようにするために使用されます。
(3) のスピンロック、つまり structnetdev_queue の _xmit_lock は、開発登録関数の相互排他的アクセス、つまりドライバーの同期を確保するために使用されます。
さらに、(1)と(3)は相互に排他的であるとカーネルコードのコメントに書かれています。(1)でロックを取得する場合は、まず(3)でロックが解放されていることを確認する必要があります。逆に、なぜこれがまだ必要なのか理解できませんでした。 。 。 。マスターがそれについて知っている場合は、アドバイスをお願いします。
3. dev_queue_xmit 関数はすでにあるのに、なぜそれを送信するためにソフト割り込みが必要なのでしょうか?
dev_queue_xmit で 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 の completed_queue に追加され、net_tx_action によってソフト割り込みの completed_queue 内のすべての skb が解放されます。
http://www.bkjia.com/PHPjc/1103186.html
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
