編輯:關於android開發
說明:本系列博文所涉及內核版本為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;
- }
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)處對應的代碼:
- static inline spinlock_t *qdisc_lock(struct Qdisc *qdisc)
- {
- return &qdisc->q.lock;
- }
所以root_lock是用於控制qdisc中skb隊列訪問的鎖,當需要對skb隊列進行enqueue、dequeue、requeue時,就需要加鎖。
__QDISC_STATE_RUNNING標志用於保證一個流控對象(qdisc)不會同時被多個cpu訪問。
而(3)處的spinlock,即structnetdev_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全部釋放掉)
開發首屏廣告(Android)簡述 作為一個成熟的應用, 必須要有廣告. 那麼, 如何優雅地開發廣告呢? 需要注意一些細節. 本文提供一個簡單的示例, 代碼僅供參考.
Android動態部署五:如何從插件apk中啟動Service 經過前面幾篇文章的分析,我們了解到了Google原生是如何拆分apk的,並且我們自己可以通過解析manif
Android 大雜燴工程之ListView的開發2以及數據倉庫開發模式 今天是放完假後工作的第一天,為了證明我不是一段幽靈代碼,我給各位看客老爺們說句:祝大家節日過得
Linux內核系列—12.c.操作系統開發之從Loader加載ELF內核,順便解釋下函數調用過程 ●,12.celf實際上,我們要做的工作是根據內核的Program he