内核是如何接受网络包的

整体一览图

image-20250310194742752

感谢张彦飞大佬的图。

在你基本上了解了什么是网卡驱动、硬中断、软中断和 ksoftirqd 线程之后，可以给出一个如上图所示的内核收包的路径示意图。

大致过程如下：

  1. 当网卡收到数据之后，以 DMA 的方式把网卡收到的帧写到内存里，再向 CPU 发起一个中断，以通知 CPU 有数据到达。
  2. 当 CPU 收到中断请求之后，会去调用网络设备驱动注册的中断处理函数。
  3. 网卡的中断处理函数并不做过多工作，发出软中断请求，然后尽快释放 CPU 资源。
  4. ksoftirqd 内核线程检测到有软中断请求到达，调用 poll 开始轮询收包，收到后交由各级协议栈处理。对于 TCP 包来说，会被放到用户 socket 的接收队列中。

做一切之前的基础准备工作

Linux 驱动、内核协议栈等模块在能够接收网卡数据包之前，要做很多的准备工作才行。如下：

  1. 提前创建好 ksoftirqd 内核线程；
  2. 要注册好各个协议对应的处理函数；
  3. 网卡设备子系统要提前初始化好；
  4. 网卡要启动好。

初始化工作

创建 ksoftirqd 内核线程

Linux 的软中断都是在专门的内核线程（ksoftirqd）中进行的，因此我们非常有必要看一下这些线程是怎么初始化的。

首先，这个线程的数量不是 1 个，而是 N 个，其中 N 等于你的机器的核数。

系统初始化的时候在 kernel/smpboot.c 中调用了 smpboot_register_percpu_thread 这个函数，该函数进一步执行到 spawn_ksoftirqd（位于 kernel/softirq.c）来创建出 softirqd 线程，执行过程如下图所示：

相关代码如下：

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)
{
    cpuhp_setup_state_nocalls(CPUHP_SOFTIRQ_DEAD, "softirq:dead", NULL,
                  takeover_tasklets);
    BUG_ON(smpboot_register_percpu_thread(&softirq_threads));


    return 0;
}
early_initcall(spawn_ksoftirqd);

当 ksoftirqd 被创建出来以后，它就会进入自己的线程循环函数 ksoftirqdshouldrun 和 run_ksoftirqd 了。接下来判断有没有软中断需要处理。

软中断不仅有网络软中断，还有其他类型。Linux 内核在 interrupt.h 中定义了所有的软中断类型，如下所示：

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


    NR_SOFTIRQS
};

网络子系统初始化

在网络子系统的初始化过程中，会为每个 CPU 初始化 softnetdata，也会为 RXSOFTIRQ 和 TX_SOFTIRQ 注册处理函数，流程如图 2.4 所示。

Linux 内核通过调用 subsys_initcall 来初始化各个子系统。

重点！！！这里说的网络子系统的初始化，会执行 netdevinit 函数。

就是这里的 subsysinitcall(netdevinit) 中的 netdev_init 函数。代码如下：

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


    /*
     *    Initialise the packet receive queues.
     */


  /*
   * 为每个 CPU 都申请一个 softnet_data 数据结构，这个数据结构里的 poll_list 用于等待驱动程序将其 poll 函数注册进来，稍后网卡驱动程序初始化的时候就可以看到这一过程了。
   */
        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；
     * NET_RX_SOFTIRQ 的处理函数为 net_rx_action；
     */
  open_softirq(NET_TX_SOFTIRQ, net_tx_action);
    open_softirq(NET_RX_SOFTIRQ, net_rx_action);
}


subsys_initcall(net_dev_init);

继续跟踪 opensoftirq 后发现这个注册的方式是记录在 softirqvec 变量里的。后面 softirqd 线程收到软中断的时候，也会使用这个变量来找到每一种软中断对应的处理函数。

void open_softirq(int nr, void (*action)(struct softirq_action *))
{
    softirq_vec[nr].action = action;
}

协议栈注册

网卡驱动初始化

每个驱动程序都会使用 module_init 向内核注册一个初始化函数，当驱动程序加载的时候，内核会调用这个函数。

调用完成后，linux 内核就会知道这个驱动的相关信息，比如 igb 网卡驱动的 igbdrivername 和 igb_probe 函数地址 等。

当网卡设备被识别之后，内核会调用其驱动的 probe 方法，（继续拿 igb 网卡驱动举例子），igbdriver 的 probe 方法是 igbprobe。

igb_probe 方法的作用就是，尽快让设备处于 ready 状态。

此外，还有一步比较关键，注册了 NAPI 机制必需的 poll 函数，这个对于 igb 网卡驱动来说，就是 igb_poll。

初始化完成之后

启动网卡

上面所有的初始化都完成以后，就可以启动网卡了。一般启动网卡的顺序都差不多，如下图所示：

igb_open 代码如下：

static int __igb_open(struct net_device *netdev, bool resuming)
{
  // 分配传输描述符数组
  err = igb_setup_all_tx_resources(adapter);
  // 分配接收描述符数组
    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));
    ......
}

igbopen 函数又调用了 igbsetupalltxresources 和 igbsetupallrxresources。在 igbsetupallrx_resources 这一步操作中，分配了 RingBuffer，并建立了内存和 Rx 队列的映射关系。

static int igb_setup_all_rx_resources(struct igb_adapter *adapter)
{
    ......


    for (i = 0; i < adapter->num_rx_queues; i++) {
        err = igb_setup_rx_resources(adapter->rx_ring[i]);
        ...
    }


    return err;
}

使用 for 循环，然后搭配 igbsetuprxresources 函数，创建了若干个队列。igbsetuprxresources 函数如下：

int igb_setup_rx_resources(struct igb_ring *rx_ring)
{
    struct device *dev = rx_ring->dev;
    int size;


    // 1. 申请 igb_rx_buffer 数组内存
    size = sizeof(struct igb_rx_buffer) * rx_ring->count;


    rx_ring->rx_buffer_info = vzalloc(size);
    if (!rx_ring->rx_buffer_info)
        goto err;


    /* Round up to nearest 4K */
    // 2. 申请 e1000_adv_rx_desc DMA 数组内存
    rx_ring->size = rx_ring->count * sizeof(union e1000_adv_rx_desc);
    rx_ring->size = ALIGN(rx_ring->size, 4096);


    rx_ring->desc = dma_alloc_coherent(dev, rx_ring->size,
                       &rx_ring->dma, GFP_KERNEL);
    if (!rx_ring->desc)
        goto err;


    // 3. 初始化队列成员
    rx_ring->next_to_alloc = 0;
    rx_ring->next_to_clean = 0;
    rx_ring->next_to_use = 0;


    return 0;


err:
    vfree(rx_ring->rx_buffer_info);
    rx_ring->rx_buffer_info = NULL;
    dev_err(dev, "Unable to allocate memory for the Rx descriptor ring\n");
    return -ENOMEM;
}

上述源码可见，实际上一个 RingBuffer 的内部不是仅有一个环形队列数组，而是有两个：

  1. igbrxbuffer 数组：这个数组是内核使用的，通过 vzalloc 申请的；
  2. e1000advrxdesc 数组：这个数组是网卡硬件使用的，通过 dmaalloc_coherent 分配。

然后其实还有最后一步中断函数的注册，注册过程看 igbrequestirq。

OK，上面就是所有的准备工作了～接下里啊就是接受数据包了。

开始接受数据包

这一部分包括了，硬中断处理

硬中断处理

首先，当数据帧从网线抵达网卡的时候，第一站是网卡的接收队列。网卡在分配给自己的 RingBuffer 中寻找可用的内存位置，找到后 DMA 引擎会将数据 DMA 到网卡之前关联的内存里，到这个时候 CPU 都是无感的。

当 DMA 操作完成后，网卡会向 CPU 发起一个硬中断，通知 CPU 有数据到达。硬中断的处理过程如下图：

在之前的“启动网卡”这一部分中，讲到网卡的硬中断注册的处理函数是 igbmsixring。

// 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;
}

其中的 igbwriteitr 只记录硬件中断频率。顺着 napischedule 调用一路跟踪下去，你就会发现，Linux 在硬中断里只完成简单必要的工作，剩下的大部分的处理都是转交给软中断的。通过以上代码可以看到，硬中断处理过程真的非常短，只是记录了一个寄存器，修改了一下 CPU 的 polllist，然后发出一个软中断，就这样，硬中断的工作就算是完成了。

ksoftirqd 内核线程处理软中断

网络包的接收处理过程主要都在 ksoftirqd 内核线程中完成，软中断都是在这里处理的，流程如下所示：

网络协议栈处理

netifreceiveskb 函数会根据包的协议进行处理，假如是 UDP 包，将包依次送到 iprcv、udprcv 等协议处理函数中进行处理。如下图：

IP 层处理

Linux 在 IP 层做的操作，在代码 net/ipv4/ip_input.c 这个代码文件中。

总结

网络模块是 Linux 内核中最复杂的模块了。整个过程，涉及到了许多内核组件之间的交互，如网卡驱动、协议栈、内核 ksoftirqd 线程等。看起来很复杂，但实际整体大概还是很清晰的。简单总结如下。

当用户执行完 recvfrom 调用之后，用户进程就通过系统调用进行到内核态工作了。如果接收队列没有数据，进程就进入睡眠状态被操作系统挂起。这块相对简单，接下来就是 LInux 各个内核组件之间的工作了。

首先在开始收包之前，Linux 要做许多的准备工作：

  • 创建 ksoftirqd 内核线程，为它设置好它自己的线程函数，后面指望着它来处理软中断；
  • 协议栈注册，Linux 要实现许多协议，比如 ARP、ICMP、IP、UDP 和 TCP，每一个协议都会将自己的处理函数注册一下，这样会方便包来了之后迅速找到对应的处理函数；
  • 网卡驱动初始化，每个驱动都有一个初始化函数，内核会让驱动也初始化一下。在这个初始化过程中，准备好自己的 DMA，并且把 NAPI 的 poll 函数地址告诉内核；
  • 启动网卡，分配 RX、TX 队列，注册中断对应的处理函数。

准备工作完成之后，接下来就是数据到来。第一个迎接它的是网卡：

  • 网卡将数据帧 DMA 到内存的 RingBUffer 中，然后向 CPU 发起中断通知；
  • CPU 响应中断请求，调用网卡启动时注册的中断处理函数；
  • 中断处理函数只是发起了软中断请求，其他的什么也没有干；
  • 内核线程 ksoftirqd 发现有软中断请求到来，先关闭硬中断；
  • ksoftirqd 线程开始调用驱动的 poll 函数收包；
  • poll 函数将收到的包送到协议栈注册的 ip_rcv 函数中；
  • iprcv 函数将包送到 udprcv 函数中（对于 TCP 包是送到 tcprcvv4）。

一些总结

问题一：RingBuffer 究竟是什么，为什么 RingBuffer 会丢包？

RingBuffer 是内存中特殊的一块区域，是一种环形队列数组，事实上这个数据结构包括了 igbrxbuffer 环形队列数组、e1000advrx_desc 环形队列数组及众多的 skb。

如果 RingBuffer 代表的是指针数组，那么是预先分配好的，如果是 skb，那么是随着收包过程而动态申请的。

问题二：软中断和硬中断分别是什么？

Linux 网络栈中数据包接收的关键流程：

  1. 硬件阶段：网卡将接收到的数据包放入 RingBuffer
  2. 硬中断触发：网卡产生硬中断通知 CPU
  3. 硬中断处理：添加网卡设备到 softnet_data 结构的 poll_list 双向链表
  4. 软中断触发：触发 NET_RX_SOFTIRQ 软中断
  5. 软中断处理：遍历 poll_list 列表，执行网卡驱动的 poll 函数收取网络包
  6. 协议栈处理：将数据包转发到 ip_rcv、udp_rcv、tcp_rcv_v4 等协议处理函数

这描述的是 Linux NAPI (New API) 机制，一种高效处理网络数据包的方法。

RingBuffer 在网络栈中的实际应用

网卡 RX/TX 环形缓冲区
/* 简化的网卡 RX Ring 结构 */
struct e1000_rx_desc {
    __le64 buffer_addr;    /* 数据缓冲区地址 */
    __le16 length;         /* 数据包长度 */
    __le16 checksum;       /* 校验和 */
    __u8  status;          /* 描述符状态 */
    __u8  errors;          /* 错误码 */
    __le16 special;
};

实际上，一个 Intel 网卡的 RX Ring 可能包含 256 个这样的描述符，形成一个环形结构。

硬中断与软中断协作的实际例子

在 Intel 82599 网卡的驱动中：

/* 硬中断处理程序 */
static irqreturn_t ixgbe_msix_lsc(int irq, void *data)
{
    struct net_device *netdev = data;


    /* 禁用网卡中断 */
    ixgbe_disable_interrupt();


    /* 将设备添加到 poll_list */
    napi_schedule(&adapter->q_vector[vector]->napi);


    return IRQ_HANDLED;
}


/* NAPI poll 函数 */
static int ixgbe_poll(struct napi_struct *napi, int budget)
{
    struct ixgbe_q_vector *q_vector = container_of(napi, struct ixgbe_q_vector, napi);
    struct ixgbe_adapter *adapter = q_vector->adapter;
    int work_done = 0;


    /* 从 RingBuffer 中批量收包，最多处理 budget 个 */
    work_done = ixgbe_clean_rx_irq(q_vector, budget);


    /* 如果工作未完成，保持在 poll_list 中 */
    if (work_done < budget) {
        napi_complete(napi);
        ixgbe_enable_interrupt();
    }


    return work_done;
}
← git 推送大文件 repo
网关性能瓶颈分析与优化技术（私藏） →