汀的知识碎片

3. 第二章 内核是如何接收网络包的

46 min read

第二章 内核是如何接收网络包的

2.1 相关实际问题

在现在的互联网的世界里，所有技术技术岗位的同学几乎都是天天和网络请求打交道。平时我们在做网络开发的时候，如果需要接收网络数据，只需要简单的几行代码就可以搞定。如果拿 C 语言来举例（Java、Golang、PHP 等其它语言也是类似的），一行 read 函数调用代码就能接收到来自对端的数据。

在开发视角看的时候，只要客户端有对应的数据发送过来，服务器端执行 read 后就能收到它。那你是否有深入思考过，在 Linux 下数据是如何从网卡一步步地到达你的进程里的，这中间都需要哪几个内核组件进行协同？这个问题看起来简单，但实际上隐藏了非常多的技术点。

1）RingBuffer 到底是啥，RingBuffer 为啥会丢包？ 在网络性能相关的技术文章中经常能看到 RingBuffer 这一关键词。RingBuffer 到底存在于哪一块，如何被用到的，真的就只是一个环形的队列吗？在技术文章里有的人说 RingBuffer 内存是预先分配好的，还有的人说 RingBuffer 里使用的内存是随着网络包的收发而动态分配的。这两个说法哪一个是正确的？为啥 RingBuffer 会丢包，如果丢包了的话应该怎么去解决？

2）网络相关的硬中断、软中断都是啥？ 听说网卡是通过硬中断来通知 CPU 有新包到达的，又听说网络里面还有个软中断。那硬中断和软中断的区别是啥，这二位又是怎么协作的呢？另外在很多性能优化的技术文章中会提到网卡中断绑定，不知道你有没有思考过为什么大家文章中都是操作的硬中断号和 CPU 之间的绑定关系。但最终的效果确实软中断跟着一起调整了，软中断开销也被绑定到了不同的 CPU 上。你有想过这是为啥吗？

3）Linux 里的 ksoftirqd 内核线程是干啥的？ 你到你的服务器上执行一下 ps -ef | grep ksoftirqd，看是不是有几个名字叫做 ksoftirqd/* 的内核线程。我把我手头虚机上的结果展示一下。

root         3     2  0 Jan04 ?        00:00:19 [ksoftirqd/0]
root        13     2  0 Jan04 ?        00:00:47 [ksoftirqd/1]
root        18     2  0 Jan04 ?        00:00:10 [ksoftirqd/2]
root        23     2  0 Jan04 ?        00:00:51 [ksoftirqd/3]

你知道这几个内核线程是做什么用的吗？你的机器上有几个，为什么有这么多。它和软中断又是什么关系？

4）为啥网卡开启多队列能提升网络性能？

int main(){
  int sock = socket(AF_INET, SOCK_STREAM, 0);
  connect(sock, ...);
  read(sock, buffer, sizeof(buffer)-1); 
  ...
}

相信不少同学在网络性能优化里听说过用多队列网卡来提升网络性能。但是你是否清楚这一性能优化方案的基本原理是什么？理解了原理你也就知道什么时候该动用这个方法，用的话开到几个合适。

5）tcpdump 是如何工作的 我们平时工作中经常会用到 tcpdump，但你知道它是如何工作的，如何和内核进行配合的么？

6）iptable/netfilter 是在哪一层实现的？ 在网络包的收发过程中，我们可以通过 iptable/netfilter 配置一些规则来进行包的过滤。那么你知道它工作在内核中的哪一层吗？

7）tcpdump 能否抓到被 iptable 封禁的包？ 如果某些数据包被 iptables 封禁，是否可以通过 tcpdump 抓到？

8）网络接收过程中的 CPU 开销如何查看？ 网络接收过程中是如何消耗 CPU 的，CPU 中的 si、sy 开销究竟是什么含义？

9）DPDK 是神马神器？ 老的还没学完，又有许多新技术出来了。比如 DPDK 究竟是什么，是否需要学习和使用它。其实你理解不了这个新技术的根本原因可能是你对 Linux kernel 工作原理不清楚。当你掌握了 Linux 内核的网络处理过程以后，回头再看 DPDK 这类 Kernel ByPass 的技术的时候，直接就有四五成的把握了。

大家可以看到，上面几个问题总体上来说都很底层。我们为什么要了解这么底层呢？如果你负责的应用不是高并发的，流量也不大，确实没有必要往下看。但是在今天的互联网公司里，由于我国人口基数大，几乎随便一个二线 App 可能都需要为百万，千万甚至几亿的用户提供稳定的的服务。深入理解 Linux 系统内部是如何实现的，以及各个部分之间是如何交互对你进行线上问题的处理，性能分析和优化将会有非常大的帮助。

带着这些疑问，让我们开始进入到网络包接收过程的探寻之旅吧！

2.2 数据是如何从网卡到协议栈的

我们在应用层执行 read 调用后就能很方便地接收到来自网络的另一端发送过来的数据。但其实在这一行代码下掩盖了非常多的内核组件的工作细节。在本小节中我们来详细看下包是如何从网卡里跑到协议栈的。另外说明一下本文网卡驱动采用 Intel 的 igb 网卡举例，其它类型的网卡工作过程类似。

2.2.1 Linux 网络收包总览

在 TCP/IP 网络分层模型里，整个协议栈被分成了物理层、链路层、网络层，传输层和应用层。应用层对应的是我们常见的 Nginx，FTP 等各种应用，也包括我们写的各种服务。Linux 内核以及网卡驱动主要实现链路层、网络层和传输层这三层。内核对更上层的应用层提供 socket 接口来供用户进程访问。我们用 Linux 的视角来看到的 TCP/IP 网络分层模型应该是如图 2.1 这样的。

图2.1 TCP/IP 网络分层模型

（此处为图2.1的文本描述：应用层 → Socket接口 → 传输层(TCP/UDP) → 网络层(IP) → 链路层(网卡驱动) → 物理层(网卡硬件)）

在 Linux 的源代码中，网络设备驱动对应的逻辑位于 driver/net/ethernet，其中 intel 系列网卡的驱动在 driver/net/ethernet/intel 目录下。协议栈模块代码位于 kernel 和 net 目录。

内核和网络设备驱动是通过中断的方式来处理的。当设备上有数据到达的时候，会给 CPU 的相关引脚上触发一个电压变化，以通知 CPU 来处理数据。对于网络模块来说，由于处理过程比较复杂和耗时，如果在中断函数中完成所有的处理，将会导致中断处理函数（优先级过高）过度占据 CPU，使得 CPU 无法响应其它设备，例如鼠标和键盘的消息。因此 Linux 中断处理函数是分上半部和下半部的。上半部是只进行最简单的工作，快速处理然后释放 CPU，接着 CPU 就可以允许其它中断进来。剩下将绝大部分的工作都放到下半部中，可以慢慢从容处理。2.4 以后的内核版本采用的下半部实现方式是软中断，由 ksoftirqd 内核线程全权处理。硬中断是通过给 CPU 物理引脚施加电压变化，而软中断是通过给内存中的一个变量的二进制值以标记有软中断发生。

好了，大概了解了网卡驱动、硬中断、软中断和 ksoftirqd 线程之后，我们在这几个概念的基础上给出一个内核收包的路径示意，如图 2.2。

图2.2 内核收包路径

（此处为图2.2的文本描述：网卡 → DMA → 内存 → 硬中断 → CPU → 软中断请求 → ksoftirqd → poll轮询 → 协议栈 → socket接收队列）

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

我们从上面这张图中已经从整体上把握到了 Linux 对数据包的处理过程。但是要想了解更多网络模块工作的细节，我们还得往下看。

2.2.2 Linux 启动

Linux 驱动，内核协议栈等等模块在具备接收网卡数据包之前，要做很多的准备工作才行。比如要提前创建好 ksoftirqd 内核线程，要注册好各个协议对应的处理函数，网卡设备子系统要提前初始化好，网卡要启动好。只有这些都 ready 之后，我们才能真正开始接收数据包。那么我们现在来看看这些准备工作都是怎么做的。

创建 ksoftirqd 内核线程

Linux 的软中断都是在专门的内核线程（ksoftirqd）中进行的，因此我们非常有必要看一下这些进程是怎么初始化的，这样我们才能在后面更准确地了解收包过程。该进程数量不是 1 个，而是 N 个，其中 N 等于你的机器的核数。

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

图2.3 创建 ksoftirqd

（此处为图2.3的文本描述：smpboot_register_percpu_thread → spawn_ksoftirqd → 创建 ksoftirqd 内核线程）

相关代码如下：

//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 了。判断有没有软中断需要被处理。这里需要注意的一点是，软中断不仅仅只有网络软中断，还有其它类型。Linux 内核在 interrupt.h 中定义了所有的软中断类型，如下所示：

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

网络子系统初始化

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

图2.4 网络子系统初始化

（此处为图2.4的文本描述：net_dev_init → 为每个CPU申请 softnet_data → open_softirq(NET_TX_SOFTIRQ, net_tx_action) → open_softirq(NET_RX_SOFTIRQ, net_rx_action)）

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

协议栈注册

内核实现了网络层的 ip 协议，也实现了传输层的 tcp 协议和 udp 协议。这些协议对应的实现函数分别是 ip_rcv()，tcp_v4_rcv() 和 udp_rcv()。和我们平时写代码的方式不一样的是，内核是通过注册的方式来实现的。

Linux 内核中的 fs_initcall 和 subsys_initcall 类似，也是初始化模块的入口。fs_initcall 调用 inet_init 后开始网络协议栈注册。通过 inet_init，将这些函数注册到了 inet_protos 和 ptype_base 数据结构中了，如图 2.5。

图2.5 协议栈注册

（此处为图2.5的文本描述：inet_init → inet_add_protocol 注册TCP/UDP到 inet_protos → dev_add_pack(&ip_packet_type) 注册IP到 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 被初始化了进来。

//file: net/ipv4/protocol.c
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 和 udp 对应的处理函数都注册到了 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 记录着 udp，tcp 的处理函数地址，ptype_base 存储着 ip_rcv() 函数的处理地址。后面我们会看到软中断中会通过 ptype_base 找到 ip_rcv 函数地址，进而将 ip 包正确地送到 ip_rcv() 中执行。

在 ip_rcv 中将会在 ip_rcv 中将会通过 inet_protos 找到 tcp 或者 udp 的处理函数，再而把包转发给 udp_rcv() 或 tcp_v4_rcv() 函数。建议大家好好读一下 inet_init 这个函数的代码。

扩展一下，如果看一下 ip_rcv 和 udp_rcv 等函数的代码能看到很多协议的处理过程。例如，ip_rcv 中会处理 iptable netfilter 过滤。再例如，udp_rcv 中会判断 socket 接收队列是否满了。对应的相关内核参数是 net.core.rmem_max 和 net.core.rmem_default。

网卡驱动初始化

每一个驱动程序（不仅仅只是网卡驱动）会使用 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 下。函数 igb_probe 主要执行的操作如图 2.6。

图2.6 网卡驱动初始化

（此处为图2.6的文本描述：igb_probe 执行步骤：1. 初始化PCI设备 → 2. 映射MMIO空间 → 3. 分配网络设备结构体 → 4. 初始化硬件（复位、设置MAC地址等） → 5. 注册ethtool回调函数 → 6. 注册net_device_ops → 7. 注册NAPI poll函数(igb_poll) → 8. 开启硬件中断）

第 5 步中我们看到，网卡驱动实现了 ethtool 所需要的接口，也在这里注册完成函数地址的注册。当 ethtool 发起一个系统调用之后，内核会找到对应操作的回调函数。对于 igb 网卡来说，其实现函数都在 drivers/net/ethernet/intel/igb/igb_ethtool.c 下。相信你这次能彻底理解 ethtool 的工作原理了吧？这个命令之所以能查看网卡收发包统计、能修改网卡自适应模式、能调整 RX 队列的数量和大小，是因为 ethtool 命令最终调用到了网卡驱动的相应方法，而不是 ethtool 本身有这个超能力。

第 6 步注册 net_device_ops 用的是 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，如下代码所示。

//file: drivers/net/ethernet/intel/igb/igb_main.c
static int igb_alloc_q_vector(...)
{
  ......
  /* initialize NAPI */
  netif_napi_add(adapter->netdev, &q_vector->napi,
                 igb_poll, 64);
}

启动网卡

当上面的初始化都完成以后，就可以启动网卡了。回忆前面网卡驱动初始化时，我们提到了驱动向内核注册了 structure net_device_ops 变量，它包含着网卡启用、发包、设置 mac 地址等回调函数（函数指针）。当启用一个网卡时（例如，通过 ifconfig eth0 up），net_device_ops 变量中定义的 ndo_open 方法会被调用。这是一个函数指针，对于 igb 网卡来说，该指针指向的是 igb_open 方法。它通常会做如图 2.7 所示的事情。

图2.7 启动网卡

（此处为图2.7的文本描述：igb_open → igb_setup_all_tx_resources（分配传输描述符数组） → igb_setup_all_rx_resources（分配接收描述符数组） → 注册中断处理函数 → 开启NAPI → 开启硬件中断）

我们来看下源码：

//file: drivers/net/ethernet/intel/igb/igb_main.c
static int __igb_open(struct net_device *netdev, bool resuming)
{
  // 分配传输描述符数组
  err = igb_setup_all_tx_resources(adapter);
  
  // 分配接收描述符数组
  err = igb_setup_all_rx_resources(adapter);
  
  ......
}

在上面的 __igb_open 函数调用了 igb_setup_all_tx_resources 和 igb_setup_all_rx_resources。在 igb_setup_all_rx_resources 这一步操作中，分配了 RingBuffer，并建立内存和 Rx 队列的映射关系。（Rx Tx 队列的数量和大小可以通过 ethtool 进行配置）。

图2.8 接收队列

（此处为图2.8的文本描述：igb_setup_all_rx_resources → 循环创建若干个接收队列(Rx Queue 0, 1, 2, …) → 每个队列包含RingBuffer(描述符数组)和DMA映射）

在上面的源码中，通过循环创建除了若干个接收队列，如图 2.8。

我们再来看一下，每一个队列是如何创建出来的。

3. 第二章 内核是如何接收网络包的

2.2.2 接收队列的创建

在上面的 __igb_open 函数调用了 igb_setup_all_tx_resources 和 igb_setup_all_rx_resources。在 igb_setup_all_rx_resources 这一步操作中，分配了 RingBuffer，并建立内存和 Rx 队列的映射关系。（Rx/Tx 队列的数量和大小可以通过 ethtool 进行配置）。

在上面的源码中，通过循环创建了若干个接收队列，如图 2.8。

图2.8 接收队列创建

我们再来看一下，每一个队列是如何创建出来的。

// 分配接收描述符数组
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));
......
}
//file: drivers/net/ethernet/intel/igb/igb_main.c
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;
}
//file: drivers/net/ethernet/intel/igb/igb_main.c
int igb_setup_tx_resources(struct igb_ring *tx_ring)
{
    //1. 申请 igb_rx_buffer 数组内存
    size = sizeof(struct igb_rx_buffer) * rx_ring->count;
    rx_ring->rx_buffer_info = vzalloc(size);
 
    //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);
 
    //3. 初始化队列成员
    rx_ring->next_to_alloc = 0;
    rx_ring->next_to_clean = 0;
    rx_ring->next_to_use = 0;
    return 0;
}

图2.9 接收队列内部

RingBuffer 内部结构

从上述源码可以看到，实际上一个 RingBuffer 的内部不仅仅是一个环形队列数组，而是有两个，如图 2.9：

  1. igb_rx_buffer 数组：这个数组是内核使用的，通过 vzalloc 申请的。
  2. e1000_adv_rx_desc 数组：这个数组是网卡硬件使用的，通过 dma_alloc_coherent 分配。

中断处理函数的注册

我们再接着看中断处理函数是如何注册的，注册过程见 igb_request_irq：

//file: drivers/net/ethernet/intel/igb/igb_main.c
static int igb_request_irq(struct igb_adapter *adapter)
{
    if (adapter->msix_entries) {
        err = igb_request_msix(adapter);
        if (!err)
            goto request_done;
        ......
    }
}

在上面的代码中跟踪函数调用：__igb_open ⇒ igb_request_irq ⇒ igb_request_msix。在 igb_request_msix 中我们看到了，对于多队列的网卡，为每一个队列都注册了中断，其对应的中断处理函数是 igb_msix_ring（该函数也在 drivers/net/ethernet/intel/igb/igb_main.c 下）。

MSI-X 中断与 CPU 绑定

我们也可以看到，MSI-X 方式下，每个 RX 队列有独立的 MSI-X 中断，从网卡硬件中断的层面就可以设置让收到的包被不同的 CPU 处理。（可以通过 irqbalance，或者修改 /proc/irq/IRQ_NUMBER/smp_affinity 来修改和 CPU 的绑定行为）。

当做好以上准备工作以后，就可以开门迎客（数据包）了！

//file: drivers/net/ethernet/intel/igb/igb_main.c
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,
                          q_vector);
    }
}

2.2.3 迎接数据的到来

硬中断处理

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

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

图2.10 硬中断处理

RingBuffer 满时的丢包

当 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：

//file: net/core/dev.c
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，这个所谓的触发过程只是对一个变量进行了一次或运算而已。

//file: kernel/softirq.c
void __raise_softirq_irqoff(unsigned int nr)
{
    trace_softirq_raise(nr);
    or_softirq_pending(1UL << nr);
}
 
//file: include/linux/interrupt.h
#define or_softirq_pending(x)  (local_softirq_pending() |= (x))
 
//file: include/linux/irq_cpustat.h
#define local_softirq_pending() \
    __IRQ_STAT(smp_processor_id(), __softirq_pending)

硬中断的职责

我们说过，Linux 在硬中断里只完成简单必要的工作，剩下的大部分处理都是转交给软中断的。通过上面代码可以看到，硬中断处理过程真的是非常短。只是记录了一个寄存器，修改了一下 CPU 的 poll_list，然后发出个软中断。就这么简单，硬中断工作就算是完成了。

ksoftirqd 内核线程处理软中断

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

图2.11 软中断处理

内核线程初始化的时候，我们介绍了 ksoftirqd 中两个线程函数 ksoftirqd_should_run 和 run_ksoftirqd。其中 ksoftirqd_should_run 代码如下：

//file: kernel/softirq.c
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 处理：

//file: kernel/softirq.c
static void run_ksoftirqd(unsigned int cpu)
{
    local_irq_disable();
    if (local_softirq_pending()) {
        __do_softirq();
        ...
    }
    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);
}

硬中断与软中断的 CPU 亲和性

这里需要注意一个细节，硬中断中设置软中断标记，和 ksoftirq 的判断是否有软中断到达，都是基于 smp_processor_id() 的。这意味着只要硬中断在哪个 CPU 上被响应，那么软中断也是在这个 CPU 上处理的。所以说，如果你发现你的 Linux 软中断 CPU 消耗都集中在一个核上的话，做法是要调整硬中断的 CPU 亲和性，来将硬中断打散到不同的 CPU 核上去。看到这里我就弄清楚了本章开头中提到的这个疑惑。

我们再来把精力集中到这个核心函数 net_rx_action 上来。

poll_list 不会重复添加

有同学问在硬中断中添加设备到 poll_list，会不会重复添加呢？答案是不会的，在软中断处理函数 net_rx_action 这里一进来就调用 local_irq_disable 把所有的硬中断都给关了，不会让硬中断重复添加 poll_list 的机会。在硬中断的处理函数中本身也有类似的判断机制，打磨了几十年的内核考虑在细节上还是很完善的。

//file: net/core/dev.c
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;
        ...
    }
}

budget 和 time_limit

函数开头的 time_limit 和 budget 是用来控制 net_rx_action 函数主动退出的，目的是保证网络包的接收不霸占 CPU 不放。等下次网卡再有硬中断过来的时候再处理剩下的接收数据包。其中 budget 可以通过内核参数调整。

这个函数中剩下的核心逻辑是获取到当前 CPU 变量 softnet_data，对其 poll_list 进行遍历，然后执行到网卡驱动注册到的 poll 函数。对于 igb 网卡来说，就是 igb 驱动里的 igb_poll 函数了。

//file: drivers/net/ethernet/intel/igb/igb_main.c
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 的调用。

//file: drivers/net/ethernet/intel/igb/igb_main.c
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 上取下来。

skb 的回收与分配

Skb 被从 RingBuffer 上取下来以后，会通过 igb_alloc_rx_buffers 申请新的 skb 再重新挂上去。所以不用担心后面新包到来的时候没有 skb 可用。

为什么需要两个函数呢？因为有可能帧要占多个 RingBuffer，所以是在一个循环中获取的，直到帧尾部。获取下来的一个数据帧用一个 sk_buff 来表示。收取完数据以后，对其进行一些校验，然后开始设置 skb 变量的 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 中，数据包将被送到协议栈中。

网络协议栈处理

netif_receive_skb 函数会根据包的协议，假如是 UDP 包，会将包依次送到 ip_rcv，udp_rcv 等协议处理函数中进行处理，如图 2.12 所示。

图2.12 网络协议栈处理

//file: net/core/dev.c
int netif_receive_skb(struct sk_buff *skb)
{
    // RPS处理逻辑,先忽略
    ......
    return __netif_receive_skb(skb);
}

在 __netif_receive_skb_core 中，我看到了原来经常使用的 tcpdump 的抓包点。

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```c
    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;
        }
    }
}

Tcpdump 是通过虚拟协议的方式工作的,它会将抓包函数以协议的形式挂到 ptype_all 上.设备层遍历所有的”协议”,这样就能抓到数据包来供我们查看了.Tcpdump 会执行到 packet_create.

//file: net/packet/af_packet.c
static int packet_create(struct net *net, struct socket *sock, ...)
{
    ...
    po->prot_hook.func = packet_rcv;
    if (sock->type == SOCK_PACKET)
        po->prot_hook.func = packet_rcv_spkt;
    po->prot_hook.af_packet_priv = sk;
    register_prot_hook(sk);
}

register_prot_hook 中会把 tcpdump 用到的”协议”挂到 ptype_all 上.看到这里很是激动,看来读一遍源代码时间真的没白浪费.

接着 __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).

IP 协议层处理

我们再来大致看一下 Linux 在 IP 协议层都做了什么,包又是怎么样进一步被送到 UDP 或 TCP 协议处理函数中的.下面是 IP 层接收网络包的主入口 ip_rcv,我们来看下.

//file: net/ipv4/ip_input.c
int ip_rcv(struct sk_buff *skb, ...)
{
    ......
    return NF_HOOK(NFPROTO_IPV4, NF_INET_PRE_ROUTING, skb, dev, NULL,
                   ip_rcv_finish);
}

Netfilter 性能影响

这里的 NF_HOOK 是一个钩子函数,它就是我们在日常工作中经常用到的 iptables/netfilter 过滤.如果你有很多或者很复杂的 netfilter 规则,会在这里消耗过多的 CPU 加大网络延迟.另外使用 NF_HOOK 在源码中搜索可以搜到很多 filter 的过滤点,想深入研究 netfilter 可以通过搜索 NF_HOOK 的这些引用处入手.通过搜索结果可以看到,主要是在 IP、ARP 等层.

当执行完注册的钩子后就会执行到最后一个参数指向的函数 ip_rcv_finish.

跟踪 ip_route_input_noref 后看到它又调用了 ip_route_input_mc.在 ip_route_input_mc 中,函数 ip_local_deliver 被赋值给了 dst.input,如下:

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

第二章 内核是如何接收网络包的

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

当执行完注册的钩子后就会执行到最后一个参数指向的函数 ip_rcv_finish.

跟踪 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).

//file: include/net/dst.h
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)
{
  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);
}

如协议注册小节看到 inet_protos 中保存着 tcp_v4_rcv 和 udp_rcv 的函数地址.这里将会根据包中的协议类型选择进行分发,在这里 skb 包将会进一步被派送到更上层的协议中,udp 和 tcp.

//file: net/ipv4/ip_input.c
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);
  }
}

协议分发

根据IP头部的协议号,从 inet_protos 数组中取出对应的处理函数(如 tcp_v4_rcv 或 udp_rcv),然后将数据包派送到更上层的协议处理.

2.2.4 收包总结

网络模块是 Linux 内核中最复杂的模块之一,看起来一个简简单单的收包过程就涉及到许多内核组件之间的交互,如网卡驱动、协议栈、内核 ksoftirqd 线程等.

看起来很复杂,本文想通过源码 + 图示的方式,尽量以容易理解的方式来将内核收包过程讲清楚.现在让我们再串一串整个收包过程.

当用户执行完 recvfrom 调用后,用户进程就通过系统调用进入到内核态工作了.如果接收队列没有数据,进程就进入睡眠状态被操作系统挂起.这块相对比较简单,剩下大部分的戏份都是由 Linux 内核其它模块来表演了.

准备工作

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

  1. 创建 ksoftirqd 线程,为它设置好它自己的线程函数,后面就指望着它来处理软中断呢.
  2. 协议栈注册,Linux 要实现许多协议,比如 arp,icmp,ip,udp,tcp,每一个协议都会将自己的处理函数注册一下,方便包来了迅速找到对应的处理函数.
  3. 网卡驱动初始化,每个驱动都有一个初始化函数,内核会让驱动也初始化一下.在这个初始化过程中,把自己的 DMA 准备好,把 NAPI 的 poll 函数地址告诉内核.
  4. 启动网卡,分配 RX、TX 队列,注册中断对应的处理函数.

以上是内核准备收包之前的重要工作,当上面都 ready 之后,就可以打开硬中断,等待数据包的到来了.

收包流程

当数据到来以后,第一个迎接它的是网卡:

  1. 网卡将数据帧 DMA 到内存的 RingBuffer 中,然后向 CPU 发起中断通知.
  2. CPU 响应中断请求,调用网卡启动时注册的中断处理函数.
  3. 中断处理函数几乎没干啥,就发起了软中断请求.
  4. 内核线程 ksoftirqd 线程发现有软中断请求到来,先关闭硬中断.
  5. ksoftirqd 线程开始调用驱动的 poll 函数收包.
  6. poll 函数将收到的包送到协议栈注册的 ip_rcv 函数中.
  7. ip_rcv 函数再将包送到 udp_rcv 函数中(对于 tcp 包是送到 tcp_rcv_v4).

收包核心路径

网卡 → DMA → RingBuffer → 硬中断 → 软中断 → ksoftirqd → poll → ip_rcv → udp_rcv/tcp_rcv_v4 → 接收缓存区

NF_HOOK 使用示例

# grep -r "NF_HOOK" *
net/ipv4/arp.c: NF_HOOK(NFPROTO_ARP, NF_ARP_OUT, skb, NULL, skb->dev, dev_queue_xmit);
net/ipv4/arp.c: return NF_HOOK(NFPROTO_ARP, NF_ARP_IN, skb, dev, NULL, arp_process);
net/ipv4/ip_input.c: return NF_HOOK(NFPROTO_IPV4, NF_INET_LOCAL_IN, skb, skb->dev, NULL, ...);
net/ipv4/ip_input.c: return NF_HOOK(NFPROTO_IPV4, NF_INET_PRE_ROUTING, skb, dev, NULL, ...);
net/ipv4/ip_forward.c: return NF_HOOK(NFPROTO_IPV4, NF_INET_FORWARD, skb, skb->dev, ...);
net/ipv4/xfrm4_output.c: return NF_HOOK_COND(NFPROTO_IPV4, NF_INET_POST_ROUTING, skb, ...);
net/ipv4/ip_output.c: NF_HOOK(NFPROTO_IPV4, NF_INET_POST_ROUTING, ...);
net/ipv4/ip_output.c: NF_HOOK(NFPROTO_IPV4, NF_INET_POST_ROUTING, newskb, ...);
net/ipv4/ip_output.c: return NF_HOOK_COND(NFPROTO_IPV4, NF_INET_POST_ROUTING, skb, NULL, ...);
net/ipv4/ip_output.c: return NF_HOOK_COND(NFPROTO_IPV4, NF_INET_POST_ROUTING, skb, NULL, ...);
......
//file: net/ipv4/ip_input.c
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);
}

2.3 本章总结

通过今天的文章，我们讲述了网络包是如何一步一步地从网卡、RingBuffer 最后到接收缓存区中的，然后内核又是如何进一步处理把它送到协议栈。理解了之后，我们回顾一下本章开头提到的几个问题。

  1. RingBuffer 到底是啥，RingBuffer 为啥会丢包？

[Image 90 on Page 21] [Image 91 on Page 21] [Image 97 on Page 22] [Image 101 on Page 23] [Image 109 on Page 25] [Image 118 on Page 27] [Image 122 on Page 28] [Image 126 on Page 29] [Image 131 on Page 30] [Image 17 on Page 31] [Image 139 on Page 32] [Image 144 on Page 33] [Image 157 on Page 37] [Image 49 on Page 42]

欢迎加入

欢迎大家加入我的知识星球，也欢迎加入我的技术交流群 Github：https://github.com/yanfeizhang/coder-kung-fu