前面和大家分享了我在CPU、記憶體、磁盤上的一點淺薄的思考，今天開始我們討論Linux里最重要的一個模塊-網路模塊，還是按照慣例來，讓我們從一段最簡單的代碼開始思考，為了簡單起見，我們用upd來舉例，如下：

int main(){
    int serverSocketFd = socket(AF_INET, SOCK_DGRAM, 0);
    bind(serverSocketFd, ...);

    char buff[BUFFSIZE];
    int readCount = recvfrom(serverSocketFd, buff, BUFFSIZE, 0, ...);
    buff[readCount] = '\0';
    printf("Receive from client:%s\n", buff);
}

上面代碼是非常簡單的一段upd server接收收據的邏輯， 當在開發視角看的時候，只要客戶端有對應的資料發送過來，服務器端執行recv_from后就能收到它，并把它列印出來，我們現在想知道的是，當網路包達到網卡，直到我們的recvfrom收到資料，這中間，究竟都發生過什么？

我們為什么要了解這么底層呢？如果你負責的應用不是高并發的，流量也不大，確實沒有必要往下看，如果你負責的是為百萬，千萬甚至過億用戶提供的服務，深入理解Linux系統內部是如何實作的，以及各個部分之間是如何互動對你的作業將會有非常大的幫助，本文基于Linux 3.10，源代碼參見https://mirrors.edge.kernel.org/pub/linux/kernel/v3.x/，網卡驅動采用Intel的igb網卡舉例，

一、Linux網路收包總覽

在TCP/IP網路分層模型里，整個協議堆疊被分成了物理層、鏈路層、網路層，傳輸層和應用層，物理層對應的是網卡和網線，應用層對應的是我們常見的Nginx，FTP等等各種應用，Linux實作的是鏈路層、網路層和傳輸層這三層，

在Linux內核實作中，鏈路層協議靠網卡驅動來實作，內核協議堆疊來實作網路層和傳輸層，內核對更上層的應用層提供socket介面來供用戶行程訪問，我們用Linux的視角來看到的TCP/IP網路分層模型應該是下面這個樣子的，

在Linux的源代碼中，網路設備驅動對應的邏輯位于driver/net/ethernet, 其中intel系列網卡的驅動在driver/net/ethernet/intel目錄下，協議堆疊模塊代碼位于kernel和net目錄，

內核和網路設備驅動是通過中斷的方式來處理的，當設備上有資料到達的時候，會給CPU的相關引腳上觸發一個電壓變化，以通知CPU來處理資料，對于網路模塊來說，由于處理程序比較復雜和耗時，如果在中斷函式中完成所有的處理，將會導致中斷處理函式（優先級過高）將過度占據CPU，將導致CPU無法回應其它設備，例如滑鼠和鍵盤的訊息，因此Linux中斷處理函式是分上半部和下半部的，上半部是只進行最簡單的作業，快速處理然后釋放CPU，接著CPU就可以允許其它中斷進來，剩下將絕大部分的作業都放到下半部中，可以慢慢從容處理，2.4以后的內核版本采用的下半部實作方式是軟中斷，由ksoftirqd內核執行緒全權處理，和硬中斷不同的是，硬中斷是通過給CPU物理引腳施加電壓變化，而軟中斷是通過給記憶體中的一個變數的二進制值以通知軟中斷處理程式，

好了，大概了解了網卡驅動、硬中斷、軟中斷和ksoftirqd執行緒之后，我們在這幾個概念的基礎上給出一個內核收包的路徑示意:

當網卡上收到資料以后，Linux中第一個作業的模塊是網路驅動， 網路驅動會以DMA的方式把網卡上收到的幀寫到記憶體里，再向CPU發起一個中斷，以通知CPU有資料到達，第二，當CPU收到中斷請求后，會去呼叫網路驅動注冊的中斷處理函式， 網卡的中斷處理函式并不做過多作業，發出軟中斷請求，然后盡快釋放CPU，ksoftirqd檢測到有軟中斷請求到達，呼叫poll開始輪詢收包，收到后交由各級協議堆疊處理，對于UPD包來說，會被放到用戶socket的接收佇列中，

我們從上面這張圖中已經從整體上把握到了Linux對資料包的處理程序，但是要想了解更多網路模塊作業的細節，我們還得往下看，

二、Linux啟動

Linux驅動，內核協議堆疊等等模塊在具備接收網卡資料包之前，要做很多的準備作業才行，比如要提前創建好ksoftirqd內核執行緒，要注冊好各個協議對應的處理函式，王闊設備子系統要提前初始化好，網卡要啟動好，只有這些都Ready之后，我們才能真正開始接收資料包，那么我們現在來看看這些準備作業都是怎么做的，

2.1 創建ksoftirqd內核行程

Linux的軟中斷都是在專門的內核執行緒（ksoftirqd）中進行的，因此我們非常有必要看一下這些行程是怎么初始化的，這樣我們才能在后面更準確地了解收包程序，該行程數量不是1個，而是N個，其中N等于你的機器的核數，

系統初始化的時候在kernel/smpboot.c中呼叫了smpboot_register_percpu_thread， 該函式進一步會執行到spawn_ksoftirqd（位于kernel/softirq.c）來創建出softirqd行程，

相關代碼如下：

//file: kernel/softirq.c
static struct smp_hotplug_thread softirq_threads = {
    .store          = &ksoftirqd,
    .thread_should_run  = ksoftirqd_should_run,
    .thread_fn      = run_ksoftirqd,
    .thread_comm        = "ksoftirqd/%u",
};

static __init int spawn_ksoftirqd(void)
{
    register_cpu_notifier(&cpu_nfb);

    BUG_ON(smpboot_register_percpu_thread(&softirq_threads));

    return 0;
}
early_initcall(spawn_ksoftirqd);

當ksoftirqd被創建出來以后，它就會進入自己的執行緒回圈函式ksoftirqd_should_run和run_ksoftirqd了，不停地判斷有沒有軟中斷需要被處理，這里需要注意的一點是，軟中斷不僅僅只有網路軟中斷，還有其它型別，

//file: include/linux/interrupt.h
enum
{
    HI_SOFTIRQ=0,
    TIMER_SOFTIRQ,
    NET_TX_SOFTIRQ,
    NET_RX_SOFTIRQ,
    BLOCK_SOFTIRQ,
    BLOCK_IOPOLL_SOFTIRQ,
    TASKLET_SOFTIRQ,
    SCHED_SOFTIRQ,
    HRTIMER_SOFTIRQ,
    RCU_SOFTIRQ,    /* Preferable RCU should always be the last softirq */

    NR_SOFTIRQS
};

2.2 網路子系統初始化

linux內核通過呼叫subsys_initcall來初始化各個子系統，在源代碼目錄里你可以grep出許多對這個函式的呼叫，這里我們要說的是網路子系統的初始化，會執行到net_dev_init函式，

//file: net/core/dev.c
static int __init net_dev_init(void)
{
    ......

    for_each_possible_cpu(i) {
        struct softnet_data *sd = &per_cpu(softnet_data, i);

        memset(sd, 0, sizeof(*sd));
        skb_queue_head_init(&sd->input_pkt_queue);
        skb_queue_head_init(&sd->process_queue);
        sd->completion_queue = NULL;
        INIT_LIST_HEAD(&sd->poll_list);

        ......
    }

    ......

    open_softirq(NET_TX_SOFTIRQ, net_tx_action);
    open_softirq(NET_RX_SOFTIRQ, net_rx_action);
}
subsys_initcall(net_dev_init);

在這個函式里，會為每個CPU都申請一個softnet_data資料結構，在這個資料結構里的poll_list是等待驅動程式將其poll函式注冊進來，稍后網卡驅動初始化的時候我們可以看到這一程序，

另外open_softirq注冊了每一種軟中斷都注冊一個處理函式， NET_TX_SOFTIRQ的處理函式為net_tx_action，NET_RX_SOFTIRQ的為net_rx_action，繼續跟蹤open_softirq后發現這個注冊的方式是記錄在softirq_vec變數里的，后面ksoftirqd執行緒收到軟中斷的時候，也會使用這個變數來找到每一種軟中斷對應的處理函式，

//file: kernel/softirq.c
void open_softirq(int nr, void (*action)(struct softirq_action *))
{
    softirq_vec[nr].action = action;
}

2.3 協議堆疊注冊

內核實作了網路層的ip協議，也實作了傳輸層的tcp協議和udp協議， 這些協議對應的實作函式分別是ip_rcv(),tcp_v4_rcv()和upd_rcv()，和我們平時寫代碼的方式不一樣的是，內核是通過注冊的方式來實作的， Linux內核中的fs_initcall和subsys_initcall類似，也是初始化模塊的入口，fs_initcall呼叫inet_init后開始網路協議堆疊注冊， 通過inet_init，將這些函式注冊到了inet_protos和ptype_base資料結構中了，如下圖:

相關代碼如下

//file: net/ipv4/af_inet.c
static struct packet_type ip_packet_type __read_mostly = {
    .type = cpu_to_be16(ETH_P_IP),
    .func = ip_rcv,
};

static const struct net_protocol udp_protocol = {
    .handler =  udp_rcv,
    .err_handler =  udp_err,
    .no_policy =    1,
    .netns_ok = 1,
};

static const struct net_protocol tcp_protocol = {
    .early_demux    =   tcp_v4_early_demux,
    .handler    =   tcp_v4_rcv,
    .err_handler    =   tcp_v4_err,
    .no_policy  =   1,
    .netns_ok   =   1,
};

static int __init inet_init(void)
{
    ......

    if (inet_add_protocol(&icmp_protocol, IPPROTO_ICMP) < 0)
        pr_crit("%s: Cannot add ICMP protocol\n", __func__);
    if (inet_add_protocol(&udp_protocol, IPPROTO_UDP) < 0)
        pr_crit("%s: Cannot add UDP protocol\n", __func__);
    if (inet_add_protocol(&tcp_protocol, IPPROTO_TCP) < 0)
        pr_crit("%s: Cannot add TCP protocol\n", __func__);

    ......

    dev_add_pack(&ip_packet_type);
}

上面的代碼中我們可以看到，udp_protocol結構體中的handler是udp_rcv，tcp_protocol結構體中的handler是tcp_v4_rcv，通過inet_add_protocol被初始化了進來，

int inet_add_protocol(const struct net_protocol *prot, unsigned char protocol)
{
    if (!prot->netns_ok) {
        pr_err("Protocol %u is not namespace aware, cannot register.\n",
            protocol);
        return -EINVAL;
    }

    return !cmpxchg((const struct net_protocol **)&inet_protos[protocol],
            NULL, prot) ? 0 : -1;
}

inet_add_protocol函式將tcp和upd對應的處理函式都注冊到了inet_protos陣列中了，再看dev_add_pack(&ip_packet_type);這一行，ip_packet_type結構體中的type是協議名，func是ip_rcv函式，在dev_add_pack中會被注冊到ptype_base哈希表中，

//file: net/core/dev.c
void dev_add_pack(struct packet_type *pt)
{
    struct list_head *head = ptype_head(pt);
    ......
}

static inline struct list_head *ptype_head(const struct packet_type *pt)
{
    if (pt->type == htons(ETH_P_ALL))
        return &ptype_all;
    else
        return &ptype_base[ntohs(pt->type) & PTYPE_HASH_MASK];
}

這里我們需要記住inet_protos記錄著upd，tcp的處理函式地址，ptype_base存盤著ip_rcv()函式的處理地址，后面我們會看到軟中斷中會通過ptype_base找到ip_rcv函式地址，進而將ip包正確地送到ip_rcv()中執行，在ip_rcv中將會通過inet_protos找到tcp或者upd的處理函式，再而把包轉發給upd_rcv()或tcp_v4_rcv()函式，

擴展一下，如果看一下ip_rcv和upd_rcv等函式的代碼能看到很多協議的處理程序，例如，ip_rcv中會處理netfilter和iptable過濾，如果你有很多或者很復雜的 netfilter 或 iptables 規則，這些規則都是在軟中斷的背景關系中執行的，會加大網路延遲，再例如，upd_rcv中會判斷socket接收佇列是否滿了，對應的相關內核引數是net.core.rmem_max和net.core.rmem_default，如果有興趣，建議大家好好讀一下inet_init這個函式的代碼，

2.4 網卡驅動初始化

每一個驅動程式（不僅僅只是網卡驅動）會使用 module_init 向內核注冊一個初始化函式，當驅動被加載時，內核會呼叫這個函式，比如igb網卡驅動的代碼位于drivers/net/ethernet/intel/igb/igb_main.c

//file: drivers/net/ethernet/intel/igb/igb_main.c
static struct pci_driver igb_driver = {
    .name     = igb_driver_name,
    .id_table = igb_pci_tbl,
    .probe    = igb_probe,
    .remove   = igb_remove,
    ......
};

static int __init igb_init_module(void)
{
    ......
    ret = pci_register_driver(&igb_driver);
    return ret;
}

驅動的pci_register_driver呼叫完成后，Linux內核就知道了該驅動的相關資訊，比如igb網卡驅動的igb_driver_name和igb_probe函式地址等等，當網卡設備被識別以后，內核會呼叫其驅動的probe方法（igb_driver的probe方法是igb_probe），驅動probe方法執行的目的就是讓設備ready，對于igb網卡，其igb_probe位于drivers/net/ethernet/intel/igb/igb_main.c下，主要執行的操作如下：

第5步中我們看到，網卡驅動實作了ethtool所需要的介面，也在這里注冊完成函式地址的注冊，當 ethtool 發起一個系統呼叫之后，內核會找到對應操作的回呼函式，對于igb網卡來說，其實作函式都在drivers/net/ethernet/intel/igb/igb_ethtool.c下， 相信你這次能徹底理解ethtool的作業原理了吧？ 這個命令之所以能查看網卡收發包統計、能修改網卡自適應模式、能調整RX 佇列的數量和大小，是因為ethtool命令最終呼叫到了網卡驅動的相應方法，而不是ethtool本身有這個超能力，

第6步注冊的igb_netdev_ops中包含的是igb_open等函式，該函式在網卡被啟動的時候會被呼叫，

//file: drivers/net/ethernet/intel/igb/igb_main.c
......
static const struct net_device_ops igb_netdev_ops = {
  .ndo_open               = igb_open,
  .ndo_stop               = igb_close,
  .ndo_start_xmit         = igb_xmit_frame,
  .ndo_get_stats64        = igb_get_stats64,
  .ndo_set_rx_mode        = igb_set_rx_mode,
  .ndo_set_mac_address    = igb_set_mac,
  .ndo_change_mtu         = igb_change_mtu,
  .ndo_do_ioctl           = igb_ioctl,......

第7步中，在igb_probe初始化程序中，還呼叫到了igb_alloc_q_vector，他注冊了一個NAPI機制所必須的poll函式，對于igb網卡驅動來說，這個函式就是igb_poll,如下代碼所示，

static int igb_alloc_q_vector(struct igb_adapter *adapter,
                  int v_count, int v_idx,
                  int txr_count, int txr_idx,
                  int rxr_count, int rxr_idx)
{

    ......
    /* initialize NAPI */
    netif_napi_add(adapter->netdev, &q_vector->napi,
               igb_poll, 64);

}

2.5 啟動網卡

當上面的初始化都完成以后，就可以啟動網卡了，回憶前面網卡驅動初始化時，我們提到了驅動向內核注冊了 structure net_device_ops 變數，它包含著網卡啟用、發包、設定mac 地址等回呼函式（函式指標），當啟用一個網卡時（例如，通過 ifconfig eth0 up），net_device_ops 中的 igb_open方法會被呼叫，它通常會做以下事情：

//file: drivers/net/ethernet/intel/igb/igb_main.c
static int __igb_open(struct net_device *netdev, bool resuming)
{
    /* allocate transmit descriptors */
    err = igb_setup_all_tx_resources(adapter);

    /* allocate receive descriptors */
    err = igb_setup_all_rx_resources(adapter);
    
    /* 注冊中斷處理函式 */
    err = igb_request_irq(adapter);
    if (err)
        goto err_req_irq;

    /* 啟用NAPI */
    for (i = 0; i < adapter->num_q_vectors; i++)
        napi_enable(&(adapter->q_vector[i]->napi));

    ......
}

在上面__igb_open函式呼叫了igb_setup_all_tx_resources,和igb_setup_all_rx_resources，在igb_setup_all_rx_resources這一步操作中，分配了RingBuffer，并建立記憶體和Rx佇列的映射關系，（Rx Tx 佇列的數量和大小可以通過 ethtool 進行配置），我們再接著看中斷函式注冊igb_request_irq:

static int igb_request_irq(struct igb_adapter *adapter)
{
    if (adapter->msix_entries) {
        err = igb_request_msix(adapter);
        if (!err)
            goto request_done;
        ......
    }
}

static int igb_request_msix(struct igb_adapter *adapter)
{
    ......
    for (i = 0; i < adapter->num_q_vectors; i++) {
        ...
        err = request_irq(adapter->msix_entries[vector].vector,
                  igb_msix_ring, 0, q_vector->name,
    }

在上面的代碼中跟蹤函式呼叫， __igb_open => igb_request_irq => igb_request_msix, 在igb_request_msix中我們看到了，對于多佇列的網卡，為每一個佇列都注冊了中斷，其對應的中斷處理函式是igb_msix_ring（該函式也在drivers/net/ethernet/intel/igb/igb_main.c下）， 我們也可以看到，msix方式下，每個 RX 佇列有獨立的MSI-X 中斷，從網卡硬體中斷的層面就可以設定讓收到的包被不同的 CPU處理，（可以通過 irqbalance ，或者修改 /proc/irq/IRQ_NUMBER/smp_affinity能夠修改和CPU的系結行為），

當做好以上準備作業以后，就可以開門迎客（資料包）了！

三、迎接資料的到來

3.1 硬中斷處理

首先當資料幀從網線到達網卡上的時候，第一站是網卡的接收佇列，網卡在分配給自己的RingBuffer中尋找可用的記憶體位置，找到后DMA引擎會把資料DMA到網卡之前關聯的記憶體里，這個時候CPU都是無感的，當DMA操作完成以后，網卡會像CPU發起一個硬中斷，通知CPU有資料到達，

注意：當RingBuffer滿的時候，新來的資料包將給丟棄，ifconfig查看網卡的時候，可以里面有個overruns，表示因為環形佇列滿被丟棄的包，如果發現有丟包，可能需要通過ethtool命令來加大環形佇列的長度，

在啟動網卡一節，我們說到了網卡的硬中斷注冊的處理函式是igb_msix_ring，

//file: drivers/net/ethernet/intel/igb/igb_main.c
static irqreturn_t igb_msix_ring(int irq, void *data)
{
    struct igb_q_vector *q_vector = data;

    /* Write the ITR value calculated from the previous interrupt. */
    igb_write_itr(q_vector);

    napi_schedule(&q_vector->napi);

    return IRQ_HANDLED;
}

igb_write_itr只是記錄一下硬體中斷頻率（據說目的是在減少對CPU的中斷頻率時用到），順著napi_schedule呼叫一路跟蹤下去，__napi_schedule=>____napi_schedule

/* Called with irq disabled */
static inline void ____napi_schedule(struct softnet_data *sd,
                     struct napi_struct *napi)
{
    list_add_tail(&napi->poll_list, &sd->poll_list);
    __raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

這里我們看到，list_add_tail修改了CPU變數softnet_data里的poll_list，將驅動napi_struct傳過來的poll_list添加了進來，
其中softnet_data中的poll_list是一個雙向串列，其中的設備都帶有輸入幀等著被處理，緊接著__raise_softirq_irqoff觸發了一個軟中斷NET_RX_SOFTIRQ， 這個所謂的觸發程序只是對一個變數進行了一次或運算而已，

void __raise_softirq_irqoff(unsigned int nr)
{
    trace_softirq_raise(nr);
    or_softirq_pending(1UL << nr);
}

//file: include/linux/irq_cpustat.h
#define or_softirq_pending(x)  (local_softirq_pending() |= (x))

我們說過，Linux在硬中斷里只完成簡單必要的作業，剩下的大部分的處理都是轉交給軟中斷的，通過上面代碼可以看到，硬中斷處理程序真的是非常短，只是記錄了一個暫存器，修改了一下下CPU的poll_list，然后發出個軟中斷，就這么簡單，硬中斷作業就算是完成了，

3.2 ksoftirqd內核執行緒處理軟中斷

內核執行緒初始化的時候，我們介紹了ksoftirqd中兩個執行緒函式ksoftirqd_should_run和run_ksoftirqd，其中ksoftirqd_should_run代碼如下：

static int ksoftirqd_should_run(unsigned int cpu)
{
    return local_softirq_pending();
}

#define local_softirq_pending() \
    __IRQ_STAT(smp_processor_id(), __softirq_pending)

這里看到和硬中斷中呼叫了同一個函式local_softirq_pending，使用方式不同的是硬中斷位置是為了寫入標記，這里僅僅只是讀取，如果硬中斷中設定了NET_RX_SOFTIRQ,這里自然能讀取的到，接下來會真正進入執行緒函式中run_ksoftirqd處理：

static void run_ksoftirqd(unsigned int cpu)
{
    local_irq_disable();
    if (local_softirq_pending()) {
        __do_softirq();
        rcu_note_context_switch(cpu);
        local_irq_enable();
        cond_resched();
        return;
    }
    local_irq_enable();
}

在__do_softirq中，判斷根據當前CPU的軟中斷型別，呼叫其注冊的action方法，

asmlinkage void __do_softirq(void)
{
    do {
        if (pending & 1) {
            unsigned int vec_nr = h - softirq_vec;
            int prev_count = preempt_count();

            ...
            trace_softirq_entry(vec_nr);
            h->action(h);
            trace_softirq_exit(vec_nr);
            ...
        }
        h++;
        pending >>= 1;
    } while (pending);
}

在網路子系統初始化小節， 我們看到我們為NET_RX_SOFTIRQ注冊了處理函式net_rx_action，所以net_rx_action函式就會被執行到了，

這里需要注意一個細節，硬中斷中設定軟中斷標記，和ksoftirq的判斷是否有軟中斷到達，都是基于smp_processor_id()的，這意味著只要硬中斷在哪個CPU上被回應，那么軟中斷也是在這個CPU上處理的，所以說，如果你發現你的Linux軟中斷CPU消耗都集中在一個核上的話，做法是要把調整硬中斷的CPU親和性，來將硬中斷打散到不通的CPU核上去，

我們再來把精力集中到這個核心函式net_rx_action上來，

static void net_rx_action(struct softirq_action *h)
{
    struct softnet_data *sd = &__get_cpu_var(softnet_data);
    unsigned long time_limit = jiffies + 2;
    int budget = netdev_budget;
    void *have;

    local_irq_disable();

    while (!list_empty(&sd->poll_list)) {
        ......
        n = list_first_entry(&sd->poll_list, struct napi_struct, poll_list);

        work = 0;
        if (test_bit(NAPI_STATE_SCHED, &n->state)) {
            work = n->poll(n, weight);
            trace_napi_poll(n);
        }

        budget -= work;
    }
}

函式開頭的time_limit和budget是用來控制net_rx_action函式主動退出的，目的是保證網路包的接收不霸占CPU不放， 等下次網卡再有硬中斷過來的時候再處理剩下的接收資料包，其中budget可以通過內核引數調整， 這個函式中剩下的核心邏輯是獲取到當前CPU變數softnet_data，對其poll_list進行遍歷, 然后執行到網卡驅動注冊到的poll函式，對于igb網卡來說，就是igb驅動力的igb_poll函式了，

/**
 *  igb_poll - NAPI Rx polling callback
 *  @napi: napi polling structure
 *  @budget: count of how many packets we should handle
 **/
static int igb_poll(struct napi_struct *napi, int budget)
{
    ...
    if (q_vector->tx.ring)
        clean_complete = igb_clean_tx_irq(q_vector);

    if (q_vector->rx.ring)
        clean_complete &= igb_clean_rx_irq(q_vector, budget);
    ...
}

在讀取操作中，igb_poll的重點作業是對igb_clean_rx_irq的呼叫，

static bool igb_clean_rx_irq(struct igb_q_vector *q_vector, const int budget)
{
    ...

    do {

        /* retrieve a buffer from the ring */
        skb = igb_fetch_rx_buffer(rx_ring, rx_desc, skb);

        /* fetch next buffer in frame if non-eop */
        if (igb_is_non_eop(rx_ring, rx_desc))
            continue;
        }

        /* verify the packet layout is correct */
        if (igb_cleanup_headers(rx_ring, rx_desc, skb)) {
            skb = NULL;
            continue;
        }

        /* populate checksum, timestamp, VLAN, and protocol */
        igb_process_skb_fields(rx_ring, rx_desc, skb);

        napi_gro_receive(&q_vector->napi, skb);
}

igb_fetch_rx_buffer和igb_is_non_eop的作用就是把資料幀從RingBuffer上取下來，為什么需要兩個函式呢？因為有可能幀要占多多個RingBuffer，所以是在一個回圈中獲取的，直到幀尾部，獲取下來的一個資料幀用一個sk_buff來表示，收取完資料以后，對其進行一些校驗，然后開始設定sbk變數的timestamp, VLAN id, protocol等欄位，接下來進入到napi_gro_receive中:

//file: net/core/dev.c
gro_result_t napi_gro_receive(struct napi_struct *napi, struct sk_buff *skb)
{
    skb_gro_reset_offset(skb);

    return napi_skb_finish(dev_gro_receive(napi, skb), skb);
}

dev_gro_receive這個函式代表的是網卡GRO特性，可以簡單理解成把相關的小包合并成一個大包就行，目的是減少傳送給網路堆疊的包數，這有助于減少 CPU 的使用量，我們暫且忽略，直接看napi_skb_finish, 這個函式主要就是呼叫了netif_receive_skb，

//file: net/core/dev.c
static gro_result_t napi_skb_finish(gro_result_t ret, struct sk_buff *skb)
{
    switch (ret) {
    case GRO_NORMAL:
        if (netif_receive_skb(skb))
            ret = GRO_DROP;
        break;
    ......
}

在netif_receive_skb中，資料包將被送到協議堆疊中，

3.3 網路協議堆疊處理

netif_receive_skb函式會根據包的協議，假如是upd包，會將包依次送到ip_rcv(),upd_rcv()協議處理函式中進行處理，

//file: net/core/dev.c
int netif_receive_skb(struct sk_buff *skb)
{
    //RPS處理邏輯，先忽略
    ......

    return __netif_receive_skb(skb);
}

static int __netif_receive_skb(struct sk_buff *skb)
{
    ......   
    ret = __netif_receive_skb_core(skb, false);
}

static int __netif_receive_skb_core(struct sk_buff *skb, bool pfmemalloc)
{
    ......

    //pcap邏輯，這里會將資料送入抓包點，tcpdump就是從這個入口獲取包的
    list_for_each_entry_rcu(ptype, &ptype_all, list) {
        if (!ptype->dev || ptype->dev == skb->dev) {
            if (pt_prev)
                ret = deliver_skb(skb, pt_prev, orig_dev);
            pt_prev = ptype;
        }
    }

    ......

    list_for_each_entry_rcu(ptype,
            &ptype_base[ntohs(type) & PTYPE_HASH_MASK], list) {
        if (ptype->type == type &&
            (ptype->dev == null_or_dev || ptype->dev == skb->dev ||
             ptype->dev == orig_dev)) {
            if (pt_prev)
                ret = deliver_skb(skb, pt_prev, orig_dev);
            pt_prev = ptype;
        }
    }
}

在__netif_receive_skb_core中，我看著原來經常使用的tcpdump的抓包點，很是激動，看來讀一遍源代碼時間真的沒白浪費，接著__netif_receive_skb_core取出protocol，它會從資料包中取出協議資訊，然后遍歷注冊在這個協議上的回呼函式串列，ptype_base 是一個 hash table，在協議注冊小節我們提到過，ip_rcv 函式地址就是存在這個 hash table中的，

//file: net/core/dev.c
static inline int deliver_skb(struct sk_buff *skb,
                  struct packet_type *pt_prev,
                  struct net_device *orig_dev)
{
    ......
    return pt_prev->func(skb, skb->dev, pt_prev, orig_dev);
}

pt_prev->func這一行就呼叫到了協議層注冊的處理函式了，對于ip包來將，就會進入到ip_rcv（如果是arp包的話，會進入到arp_rcv），

3.4 IP協議層處理

我們再來大致看一下linux在ip協議層都做了什么，包又是怎么樣進一步被送到udp或tcp協議處理函式中的，

//file: net/ipv4/ip_input.c
int ip_rcv(struct sk_buff *skb, struct net_device *dev, struct packet_type *pt, struct net_device *orig_dev)
{
    ......

    return NF_HOOK(NFPROTO_IPV4, NF_INET_PRE_ROUTING, skb, dev, NULL,
               ip_rcv_finish);
}

這里NF_HOOK是一個鉤子函式，當執行完注冊的鉤子后就會執行到最后一個引數指向的函式ip_rcv_finish，

static int ip_rcv_finish(struct sk_buff *skb)
{
    ......

    if (!skb_dst(skb)) {
        int err = ip_route_input_noref(skb, iph->daddr, iph->saddr,
                           iph->tos, skb->dev);
        ...
    }

    ......

    return dst_input(skb);
}

跟蹤ip_route_input_noref 后看到它又呼叫了 ip_route_input_mc， 在ip_route_input_mc中，函式ip_local_deliver被賦值給了dst.input, 如下：

//file: net/ipv4/route.c
static int ip_route_input_mc(struct sk_buff *skb, __be32 daddr, __be32 saddr,
                u8 tos, struct net_device *dev, int our)
{
    if (our) {
        rth->dst.input= ip_local_deliver;
        rth->rt_flags |= RTCF_LOCAL;
    }
}

所以回到ip_rcv_finish中的return dst_input(skb);，

/* Input packet from network to transport.  */
static inline int dst_input(struct sk_buff *skb)
{
    return skb_dst(skb)->input(skb);
}

skb_dst(skb)->input呼叫的input方法就是路由子系統賦的ip_local_deliver，

//file: net/ipv4/ip_input.c
int ip_local_deliver(struct sk_buff *skb)
{
    /*
     *  Reassemble IP fragments.
     */

    if (ip_is_fragment(ip_hdr(skb))) {
        if (ip_defrag(skb, IP_DEFRAG_LOCAL_DELIVER))
            return 0;
    }

    return NF_HOOK(NFPROTO_IPV4, NF_INET_LOCAL_IN, skb, skb->dev, NULL,
               ip_local_deliver_finish);
}

static int ip_local_deliver_finish(struct sk_buff *skb)
{
    ......

    int protocol = ip_hdr(skb)->protocol;
    const struct net_protocol *ipprot;

    ipprot = rcu_dereference(inet_protos[protocol]);
    if (ipprot != NULL) {
        ret = ipprot->handler(skb);
    }
}

如協議注冊小節看到inet_protos中保存著tcp_rcv()和udp_rcv()的函式地址，這里將會根據包中的協議型別選擇進行分發,在這里skb包將會進一步被派送到更上層的協議中，udp和tcp，

3.5 UDP協議層處理

在協議注冊小節的時候我們說過，udp協議的處理函式是udp_rcv，

//file: net/ipv4/udp.c
int udp_rcv(struct sk_buff *skb)
{
    return __udp4_lib_rcv(skb, &udp_table, IPPROTO_UDP);
}


int __udp4_lib_rcv(struct sk_buff *skb, struct udp_table *udptable,
           int proto)
{
    sk = __udp4_lib_lookup_skb(skb, uh->source, uh->dest, udptable);

    if (sk != NULL) {
        int ret = udp_queue_rcv_skb(sk, skb
    }

    icmp_send(skb, ICMP_DEST_UNREACH, ICMP_PORT_UNREACH, 0);
}

__udp4_lib_lookup_skb是根據skb來尋找對應的socket，當找到以后將資料包放到socket的快取佇列里，如果沒有找到，則發送一個目標不可達的icmp包，

//file: net/ipv4/udp.c
int udp_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
{   
    ......

    if (sk_rcvqueues_full(sk, skb, sk->sk_rcvbuf))
        goto drop;

        rc = 0;

    ipv4_pktinfo_prepare(skb);
    bh_lock_sock(sk);
    if (!sock_owned_by_user(sk))
        rc = __udp_queue_rcv_skb(sk, skb);
    else if (sk_add_backlog(sk, skb, sk->sk_rcvbuf)) {
        bh_unlock_sock(sk);
        goto drop;
    }
    bh_unlock_sock(sk);

    return rc;
}

sock_owned_by_user判斷的是用戶是不是正在這個socker上進行系統呼叫（socket被占用），如果沒有，那就可以直接放到socket的接收佇列中，如果有，那就通過sk_add_backlog把資料包添加到backlog佇列， 當用戶釋放的socket的時候，內核會檢查backlog佇列，如果有資料再移動到接收佇列中，

sk_rcvqueues_full接收佇列如果滿了的話，將直接把包丟棄，接收佇列大小受內核引數net.core.rmem_max和net.core.rmem_default影響，

四、recvfrom系統呼叫

花開兩朵，各表一枝， 上面我們說完了整個Linux內核對資料包的接收和處理程序，最后把資料包放到socket的接收佇列中了，那么我們再回頭看用戶行程呼叫recvfrom后是發生了什么， 我們在代碼里呼叫的recvfrom是一個glibc的庫函式，該函式在執行后會將用戶進行陷入到內核態，進入到Linux實作的系統呼叫sys_recvfrom，在理解Linux對sys_recvfrom之前，我們先來簡單看一下socket這個核心資料結構，這個資料結構太大了，我們只把對和我們今天主題相關的內容畫出來，如下：

socket資料結構中的const struct proto_ops對應的是協議的方法集合，每個協議都會實作不同的方法集，對于IPv4 Internet協議族來說,每種協議都有對應的處理方法，如下，對于udp來說，是通過inet_dgram_ops來定義的，其中注冊了inet_recvmsg方法，

//file: net/ipv4/af_inet.c
const struct proto_ops inet_stream_ops = {
    ......
    .recvmsg       = inet_recvmsg,
    .mmap          = sock_no_mmap,
    ......
}
const struct proto_ops inet_dgram_ops = {
    ......
    .sendmsg       = inet_sendmsg,
    .recvmsg       = inet_recvmsg,
    ......
}

socket資料結構中的另一個資料結構struct sock *sk是一個非常大，非常重要的子結構體，其中的sk_prot又定義了二級處理函式，對于UPD協議來說，會被設定成UPD協議實作的方法集udp_prot，

//file: net/ipv4/udp.c
struct proto udp_prot = {
    .name          = "UDP",
    .owner         = THIS_MODULE,
    .close         = udp_lib_close,
    .connect       = ip4_datagram_connect,
    ......
    .sendmsg       = udp_sendmsg,
    .recvmsg       = udp_recvmsg,
    .sendpage      = udp_sendpage,
    ......
}

看完了socket變數之后，我們再來看sys_revvfrom的實作程序，

在inet_recvmsg呼叫了sk->sk_prot->recvmsg，

//file: net/ipv4/af_inet.c
int inet_recvmsg(struct kiocb *iocb, struct socket *sock, struct msghdr *msg,
         size_t size, int flags)
{   
    ......
    err = sk->sk_prot->recvmsg(iocb, sk, msg, size, flags & MSG_DONTWAIT,
                   flags & ~MSG_DONTWAIT, &addr_len);
    if (err >= 0)
        msg->msg_namelen = addr_len;
    return err;
}

上面我們說過這個對于upd協議的socket來說，這個sk_prot就是net/ipv4/udp.c下的struct proto udp_prot，由此我們找到了udp_recvmsg方法，

//file: net/core/datagram.c:EXPORT_SYMBOL(__skb_recv_datagram);
struct sk_buff *__skb_recv_datagram(struct sock *sk, unsigned int flags,
                    int *peeked, int *off, int *err)
{
    ......
    do {
        struct sk_buff_head *queue = &sk->sk_receive_queue;
        skb_queue_walk(queue, skb) {
            ......
        }

        /* User doesn't want to wait */
        error = -EAGAIN;
        if (!timeo)
            goto no_packet;
    } while (!wait_for_more_packets(sk, err, &timeo, last));
}
}

終于我們找到了我們想要看的重點，在上面我們看到了所謂的讀取程序，就是訪問sk->sk_receive_queue，如果沒有資料，且用戶也允許等待，則將呼叫wait_for_more_packets()執行等待操作，它加入會讓用戶行程進入睡眠狀態，

五、總結

網路模塊是Linux內核中最復雜的模塊了，看起來一個簡簡單單的收包程序就涉及到許多內核組件之間的互動，如網卡驅動、協議堆疊，內核ksoftirqd執行緒等，
看起來很復雜，本文想通過圖示的方式，盡量以容易理解的方式來將內核收包程序講清楚，現在讓我們再串一串整個收包程序，

當用戶執行完recvfrom呼叫后，用戶行程就通過系統呼叫進行到內核態作業了，如果接收佇列沒有資料，行程就進入睡眠狀態被作業系統掛起，這塊相對比較簡單，剩下大部分的戲份都是由Linux內核其它模塊來表演了，

首先在開始收包之前，Linux要做許多的準備作業：

• 創建ksoftirqd執行緒，為它設定好它自己的執行緒函式，后面就指望著它來處理軟中斷呢，
• 協議堆疊注冊，linux要實作許多協議，比如arp，icmp，ip，udp，tcp，每一個協議都會將自己的處理函式注冊一下，方便包來了迅速找到對應的處理函式
• 網卡驅動初始化，每個驅動都有一個初始化函式，內核會讓驅動也初始化一下，在這個初始化程序中，把自己的DMA準備好，把NAPI的poll函式地址告訴內核
• 啟動網卡，分配RX，TX佇列，注冊中斷對應的處理函式

以上是內核準備收包之前的重要作業，當上面都ready之后，就可以打開硬中斷，等待資料包的到來了，

當資料到到來了以后，第一個迎接它的是網卡（我去，這不是廢話么）：

• 網卡將資料幀DMA到記憶體的RingBuffer中，然后向CPU發起中斷通知
• CPU回應中斷請求，呼叫網卡啟動時注冊的中斷處理函式
• 中斷處理函式幾乎沒干啥，就發起了軟中斷請求
• 內核執行緒ksoftirqd執行緒發現有軟中斷請求到來，先關閉硬中斷
• ksoftirqd執行緒開始呼叫驅動的poll函式收包
• poll函式將受到的包送到協議堆疊注冊的ip_rcv函式中
• ip_rcv函式再講包送到upd_rcv函式中（對于tcp包就送到tcp_rcv）

現在我們可以回到開篇的問題了，我們在用戶層看到的簡單一行recvfrom,Linux內核要替我們做如此之多的作業，才能讓我們順利收到資料，這還是簡簡單單的UDP，如果是TCP，內核要做的作業更多，不由得感嘆內核的開發者們真的是用心良苦，

理解了整個收包程序以后，我們就能明確知道Linux收一個包的CPU開銷了，首先第一塊是用戶行程呼叫系統呼叫陷入內核態的開銷，第二塊是CPU回應包的硬中斷的CPU開銷，第三塊是ksoftirqd內核執行緒的軟中斷背景關系花費的，后面我們再專門發一篇文章實際觀察一下這些開銷，

另外網路收發中有很多末只細節咱們并沒有展開了說，比如說no NAPI， GRO，RPS等，因為我覺得說的太對了反而會影響大家對整個流程的把握，所以盡量只保留主框架了，少即是多！

開發內功修煉之硬碟篇專輯：

• 圖解Linux網路包接收程序
• Linux網路包接收程序的監控與調優
• 聊聊TCP連接耗時的那些事兒

我的公眾號是「開發內功修煉」，在這里我不是單純介紹技術理論，也不只介紹實踐經驗，而是把理論與實踐結合起來，用實踐加深對理論的理解、用理論提高你的技術實踐能力，歡迎你來關注我的公眾號，也請分享給你的好友~~~

標籤：PHP

上一篇：CoProcessFunction實戰三部曲之二：狀態處理

下一篇：DRF 自動生成介面檔案 coreapi和drf-yasg