二層（鏈路層）數據包發送過程分析

二層（鏈路層）數據包發送過程分析

說明：本系列博文所涉及內核版本為2.6.32
當上層准備好一個包之後，交給鏈路層，鏈路層數據包發送主要通過dev_queue_xmit函數處理。數據包的發送可分為兩種，一種是正常的傳輸流程，即通過網卡驅動，另一種是通過軟中斷（見注3）。為了理解方便，首先看一下dev_queue_xmi函數的整體調用關系圖。

ldev_queue_xmit

本函數用來將帶發送的skb加入一個dev的隊列（Queue），調用這個函數前必須設置好skb的device和priority，本函數可以在中斷上下文中被調用。

返回值：

返回非0（正數或負數）表示函數出錯，返回0表示成功，但是並不表示數據包被成功發送出去，因為數據包可能因為限速等原因被丟掉。

函數執行後傳入的skb將被釋放，所以如果想控制數據包，實現對skb的重傳時需要增加skb的引用計數。

當調用此函數時中斷必須是打開的，因為BHenable必須要求IRQenable，否則會造成死鎖。

int dev_queue_xmit(struct sk_buff *skb)
 
{
struct net_device *dev = skb->dev;
struct netdev_queue *txq;
struct Qdisc *q;
int rc = -ENOMEM;
/* GSO will handle the following emulations directly. */
if (netif_needs_gso(dev, skb))
goto gso;
if (skb_has_frags(skb) &&
!(dev->features & NETIF_F_FRAGLIST) &&
__skb_linearize(skb))
goto out_kfree_skb;
//如果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的線性化。
if (skb_shinfo(skb)->nr_frags &&
(!(dev->features & NETIF_F_SG) || illegal_highdma(dev, skb)) &&
__skb_linearize(skb))
goto out_kfree_skb;
//如果數據包沒有被計算校驗和並且發送設備不支持這個協議的校驗，則在此進行校驗和的計算（注1）。如果上面已經線性化了一次，這裡的__skb_linearize就會直接返回,注意區別frags和frag_list，前者是將多的數據放到單獨分配的頁面中，sk_buff只有一個。而後者則是連接多個sk_buff 
if (skb->ip_summed == CHECKSUM_PARTIAL) {
skb_set_transport_header(skb, skb->csum_start -
skb_headroom(skb));
if (!dev_can_checksum(dev, skb) && skb_checksum_help(skb))
goto out_kfree_skb;
}
gso:
//關閉軟中斷,禁止cpu搶占
rcu_read_lock_bh();
//選擇一個發送隊列，如果設備提供了select_queue回調函數就使用它，否則由內核選擇一個隊列,這裡只是Linux內核多隊列的實現，但是要真正的使用都隊列，需要網卡支持多隊列才可以，一般的網卡都只有一個隊列。在調用alloc_etherdev分配net_device是，設置隊列的個數
txq = dev_pick_tx(dev, skb);
// 從netdev_queue結構上獲取設備的qdisc 
q = rcu_dereference(txq->qdisc);
//如果該設備有隊列可用，就調用__dev_xmit_skb 
if (q->enqueue) {
rc = __dev_xmit_skb(skb, q, dev, txq);
goto out;
}
//下面的處理是在沒有發送隊列的情況,軟設備一般沒有發送隊列：如lo、tunnle；我們所要做的就是直接調用驅動的hard_start_xmit將它發送出去  如果發送失敗就直接丟棄，因為沒有隊列可以保存它 
if (dev->flags & IFF_UP) { //確定設備是否開啟
int cpu = smp_processor_id(); /* ok because BHs are off */
if (txq->xmit_lock_owner != cpu) {//是否在同一個cpu上
HARD_TX_LOCK(dev, txq, cpu);
if (!netif_tx_queue_stopped(txq)) {//確定隊列是運行狀態
rc = NET_XMIT_SUCCESS;
if (!dev_hard_start_xmit(skb, dev, txq)) {
HARD_TX_UNLOCK(dev, txq);
goto out;
}
}
HARD_TX_UNLOCK(dev, txq);
if (net_ratelimit())
printk(KERN_CRIT "Virtual device %s asks to "
"queue packet!\n", dev->name);
} else {// txq->xmit_lock_owner == cpu的情況，說明發生遞歸
if (net_ratelimit())
printk(KERN_CRIT "Dead loop on virtual device "
"%s, fix it urgently!\n", dev->name);
}
}
rc = -ENETDOWN;
rcu_read_unlock_bh();
out_kfree_skb:
kfree_skb(skb);
return rc;
out:
rcu_read_unlock_bh();
return rc;
}

l__dev_xmit_skb

__dev_xmit_skb函數主要做兩件事情：

（1）如果流控對象為空的，試圖直接發送數據包。

（2）如果流控對象不空，將數據包加入流控對象，並運行流控對象。

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);//見注2
int rc;
spin_lock(root_lock); //鎖qdisc
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)) {
/*
* 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.
*/
__qdisc_update_bstats(q, skb->len);
if (sch_direct_xmit(skb, q, dev, txq, root_lock))
__qdisc_run(q);
else
clear_bit(__QDISC_STATE_RUNNING, &q->state);
rc = NET_XMIT_SUCCESS;
} else {
rc = qdisc_enqueue_root(skb, q);
qdisc_run(q);
}
spin_unlock(root_lock);
return rc;
}

lqdisc_run

有兩個時機將會調用qdisc_run()：

1．__dev_xmit_skb()

2.軟中斷服務線程NET_TX_SOFTIRQ

static inline void qdisc_run(struct Qdisc *q)
 
{
if (!test_and_set_bit(__QDISC_STATE_RUNNING, &q->state))//將隊列設置為運行狀態
__qdisc_run(q);
}

l__qdisc_run

void __qdisc_run(struct Qdisc *q)
 
{
unsigned long start_time = jiffies;
while (qdisc_restart(q)) { //返回值大於0，說明流控對象非空
/*如果發現本隊列運行的時間太長了，將會停止隊列的運行，並將隊列加入output_queue鏈表頭
* Postpone processing if (延遲處理)
* 1. another process needs the CPU;
* 2. we've been doing it for too long.
*/
if (need_resched() || jiffies != start_time) { //已經不允許繼續運行本流控對象
__netif_schedule(q); //將本qdisc加入每cpu變量softnet_data的output_queue鏈表中
break;
}
}
//清除隊列的運行標識
clear_bit(__QDISC_STATE_RUNNING, &q->state);
}

循環調用qdisc_restart發送數據,下面這個函數qdisc_restart是真正發送數據包的函數,它從隊列上取下一個幀，然後嘗試將它發送出去,若發送失敗則一般是重新入隊。

此函數返回值為：發送成功時返回剩余隊列長度,發送失敗時返回0（若發送成功且剩余隊列長度為0也返回0）

lqdisc_restart

__QDISC_STATE_RUNNING狀態保證同一時刻只有一個cpu在處理這個qdisc，qdisc_lock(q)用來保證對這個隊列的順序訪問。

通常netif_tx_lock用來確保本設備驅動的順序（獨占）訪問的，qdisc_lock(q)用來保證qdisc的順序訪問，這兩個是互斥的，獲得其中一個必須釋放另一個。

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); //一開始就調用dequeue函數
if (unlikely(!skb))
return 0; //返回0說明隊列是空的或者被限制
root_lock = qdisc_lock(q);
dev = qdisc_dev(q);
txq = netdev_get_tx_queue(dev, skb_get_queue_mapping(skb));
return sch_direct_xmit(skb, q, dev, txq, root_lock); //用於發送數據包
}

lsch_direct_xmit

發送一個skb，將隊列置為__QDISC_STATE_RUNNING狀態，保證只有一個cpu運行這個函數，返回0表示隊列為空或者發送受限，大於0表示隊列非空。

int sch_direct_xmit(struct sk_buff *skb, struct Qdisc *q,
 
struct net_device *dev, struct netdev_queue *txq,
spinlock_t *root_lock)
{
int ret = NETDEV_TX_BUSY;
spin_unlock(root_lock);// release qdisc,因為後面要獲取設備鎖
// 調用__netif_tx_lockà spin_lock(&txq->_xmit_lock,，保證設備驅動的獨占訪問
HARD_TX_LOCK(dev, txq, smp_processor_id());
if (!netif_tx_queue_stopped(txq) && //設備沒有被停止，且發送隊列沒有被凍結
!netif_tx_queue_frozen(txq))
ret = dev_hard_start_xmit(skb, dev, txq); //發送數據包
HARD_TX_UNLOCK(dev, txq); // 調用__netif_tx_unlock
spin_lock(root_lock);
switch (ret) {
case NETDEV_TX_OK: //如果設備成功將數據包發送出去
ret = qdisc_qlen(q); //返回剩余的隊列長度
break;
case NETDEV_TX_LOCKED: //獲取設備鎖失敗
ret = handle_dev_cpu_collision(skb, txq, q);
break;
default: //設備繁忙，重新入隊發送（利用softirq）
if (unlikely (ret != NETDEV_TX_BUSY && net_ratelimit()))
printk(KERN_WARNING "BUG %s code %d qlen %d\n",
dev->name, ret, q->q.qlen);
ret = dev_requeue_skb(skb, q);
break;
}
if (ret && (netif_tx_queue_stopped(txq) ||
netif_tx_queue_frozen(txq)))
ret = 0;
return ret;
}

ldev_hard_start_xmit

int dev_hard_start_xmit(struct sk_buff *skb, struct net_device *dev,
 
struct netdev_queue *txq)
{
const struct net_device_ops *ops = dev->netdev_ops;
int rc;
if (likely(!skb->next)) {
//從這裡可以看出，對於每一個發送的包也會發給ptype_all一份，  而packet套接字創建時對於proto為ETH_P_ALL的會在ptype_all中注冊一個成員,因此對於協議號為ETH_P_ALL的packet套接字來說，發送和接受的數據都能收到
if (!list_empty(&ptype_all))
dev_queue_xmit_nit(skb, dev);
if (netif_needs_gso(dev, skb)) {
if (unlikely(dev_gso_segment(skb)))
goto out_kfree_skb;
if (skb->next)
goto gso;
}
//如果發送設備不需要skb->dst，則在此將其釋放
if (dev->priv_flags & IFF_XMIT_DST_RELEASE)
skb_dst_drop(skb);
//調用設備注冊的發送函數，即dev->netdev_ops-> ndo_start_xmit(skb, dev)
rc = ops->ndo_start_xmit(skb, dev);
if (rc == NETDEV_TX_OK)
txq_trans_update(txq);
return rc;
}
gso:
……
}

ldev_queue_xmit_nit

static void dev_queue_xmit_nit(struct sk_buff *skb, struct net_device *dev)
 
{
struct packet_type *ptype;
#ifdef CONFIG_NET_CLS_ACT
if (!(skb->tstamp.tv64 && (G_TC_FROM(skb->tc_verd) & AT_INGRESS)))
net_timestamp(skb); //記錄該數據包輸入的時間戳
#else
net_timestamp(skb);
#endif
rcu_read_lock();
list_for_each_entry_rcu(ptype, &ptype_all, list) {
/* Never send packets back to the socket they originated from */
//遍歷ptype_all鏈表，查找所有符合輸入條件的原始套接口，並循環將數據包輸入到滿足條件的套接口
if ((ptype->dev == dev || !ptype->dev) &&
(ptype->af_packet_priv == NULL ||
(struct sock *)ptype->af_packet_priv != skb->sk)) {
//由於該數據包是額外輸入到這個原始套接口的，因此需要克隆一個數據包
struct sk_buff *skb2 = skb_clone(skb, GFP_ATOMIC);
if (!skb2)
break;
/* skb->nh should be correctly(確保頭部偏移正確)
set by sender, so that the second statement is
just protection against buggy protocols.
*/
skb_reset_mac_header(skb2);
if (skb_network_header(skb2) < skb2->data ||
skb2->network_header > skb2->tail) {
if (net_ratelimit())//net_ratelimit用來保證網絡代碼中printk的頻率
printk(KERN_CRIT "protocol %04x is "
"buggy, dev %s\n",
skb2->protocol, dev->name);
skb_reset_network_header(skb2); //重新設置L3頭部偏移
}
skb2->transport_header = skb2->network_header;
skb2->pkt_type = PACKET_OUTGOING;
ptype->func(skb2, skb->dev, ptype, skb->dev);//調用協議(ptype_all)接受函數
}
}
rcu_read_unlock();
}

?環回設備

對於環回設備loopback，設備的ops->ndo_start_xmit被初始化為loopback_xmit函數。

static const struct net_device_ops loopback_ops = {
 
.ndo_init = loopback_dev_init,
.ndo_start_xmit= loopback_xmit,
.ndo_get_stats = loopback_get_stats,
};

drivers/net/loopback.c

static netdev_tx_t loopback_xmit(struct sk_buff *skb,
 
struct net_device *dev)
{
struct pcpu_lstats *pcpu_lstats, *lb_stats;
int len;
skb_orphan(skb);
skb->protocol = eth_type_trans(skb, dev);
/* it's OK to use per_cpu_ptr() because BHs are off */
pcpu_lstats = dev->ml_priv;
lb_stats = per_cpu_ptr(pcpu_lstats, smp_processor_id());
len = skb->len;
if (likely(netif_rx(skb) == NET_RX_SUCCESS)) { //直接調用了netif_rx進行了接收處理
lb_stats->bytes += len;
lb_stats->packets++;
} else
lb_stats->drops++;
return NETDEV_TX_OK;
}

• 注：

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