【Linux 网络子系统深度拆解】收包路径全解：从 NIC 中断到 socket 接收队列

你在用 perf top 看一台高流量网关的 CPU 热点，发现 net_rx_action、__netif_receive_skb_core、ip_rcv、tcp_v4_rcv 这几个函数轮流霸榜。你知道它们都在”收包路径”上，但它们之间是怎么串起来的？net_rx_action 的 budget 机制是怎么控制一次软中断最多处理多少包的？__netif_receive_skb_core 里的 ptype_all 和 ptype_base 分别是什么？tcp_v4_rcv 拿到包之后，如果 socket 正被用户态锁住，包会去哪里？

前两篇我们拆解了 sk_buff（包的容器）和 net_device / NAPI（硬件到内核的接口）。现在把它们串起来：从网卡收到一个以太网帧开始，追踪它在内核中走过的每一个函数，直到用户态的 recvmsg() 拿到数据。

本文基于 Linux 6.6 LTS 内核源码。6.8 的差异会单独标注。源码路径均相对于内核源码根目录。

一、全景图：RX 收包路径的八个阶段

先给出全貌。一个网络包从网线到用户态，经过以下八个阶段：

整条路径在同一个 CPU 核心上执行（除非启用了 RPS，后面会讨论）。这意味着收包的 CPU 开销全部集中在处理硬中断的那个核心上——这是理解网络性能瓶颈的关键前提。

二、硬中断：从 DMA 完成到 napi_schedule

2.1 DMA 写入和中断触发

网卡通过 DMA（Direct Memory Access）把收到的帧直接写入内核预分配的 ring buffer 内存区域。DMA 完成后，网卡向 CPU 发送一个 MSI-X 中断。

在多队列网卡中，每个 RX 队列有独立的 MSI-X 中断向量，可以绑定到不同的 CPU 核心。这是 RSS（Receive Side Scaling）的基础——通过硬件哈希把不同流（flow）分散到不同队列，从而分散到不同 CPU。

2.2 驱动中断处理函数

以 Intel i40e 驱动为例，中断处理函数的核心逻辑只做一件事——调度 NAPI：

// drivers/net/ethernet/intel/i40e/i40e_txrx.c
static irqreturn_t i40e_msix_clean_rings(int irq, void *data)
{
    struct i40e_q_vector *q_vector = data;

    if (!q_vector->tx.ring && !q_vector->rx.ring)
        return IRQ_HANDLED;

    napi_schedule_irqoff(&q_vector->napi);

    return IRQ_HANDLED;
}

注意这里调用的是 napi_schedule_irqoff() 而不是 napi_schedule()——因为已经在硬中断上下文中，中断已经被关闭，不需要再 local_irq_save。

2.3 napi_schedule 的状态机

napi_schedule() 不是简单地”把 NAPI 加到队列里”。它有一个基于原子位操作的状态机：

// include/linux/netdevice.h:507
bool napi_schedule_prep(struct napi_struct *n);

// include/linux/netdevice.h:519
static inline bool napi_schedule(struct napi_struct *n)
{
    if (napi_schedule_prep(n)) {
        __napi_schedule(n);
        return true;
    }
    return false;
}

napi_schedule_prep() 通过 test_and_set_bit(NAPI_STATE_SCHED, &n->state) 原子地检查并设置调度标志。如果 NAPI 已经在调度状态（上一次 poll 还没结束），直接返回 false——这是防止同一个 NAPI 被重复调度的关键机制。

__napi_schedule() 的核心操作：

// net/core/dev.c
void __napi_schedule(struct napi_struct *n)
{
    unsigned long flags;

    local_irq_save(flags);
    ____napi_schedule(this_cpu_ptr(&softnet_data), n);
    local_irq_restore(flags);
}

static inline void ____napi_schedule(struct softnet_data *sd,
                                     struct napi_struct *napi)
{
    struct task_struct *thread;

    // 如果启用了 threaded NAPI (6.1+)
    if (test_bit(NAPI_STATE_THREADED, &napi->state)) {
        // 唤醒 NAPI 线程
        thread = napi->thread;
        if (thread && thread != current)
            wake_up_process(thread);
        return;
    }

    // 否则加入 per-CPU 的 poll_list
    list_add_tail(&napi->poll_list, &sd->poll_list);
    // 触发 NET_RX_SOFTIRQ
    __raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

关键数据结构是 softnet_data——每个 CPU 一个实例：

// include/linux/netdevice.h:3281
struct softnet_data {
    struct list_head    poll_list;      // 待轮询的 NAPI 列表
    struct sk_buff_head process_queue;  // 待处理的 skb 队列
    unsigned int        processed;      // 已处理包计数
    unsigned int        time_squeeze;   // 因时间耗尽而退出的次数

    bool                in_net_rx_action;   // 正在执行 net_rx_action
    bool                in_napi_threaded_poll;

    struct sk_buff_head input_pkt_queue;    // RPS 入队队列
    struct napi_struct  backlog;            // per-CPU backlog NAPI
    unsigned int        received_rps;
    unsigned int        dropped;            // 因队列满而丢弃的计数
    // ...
};
DECLARE_PER_CPU_ALIGNED(struct softnet_data, softnet_data);

softnet_data 里的 time_squeeze 是一个关键的监控指标——它记录了 net_rx_action 因为预算或时间耗尽而被迫退出的次数。如果这个值持续增长，说明 CPU 来不及处理所有收到的包。

三、net_rx_action：软中断的预算游戏

当 NET_RX_SOFTIRQ 被触发后，内核的软中断处理路径最终会调用 net_rx_action()。这是整个收包路径的调度中枢。

3.1 预算模型

net_rx_action 有两个退出条件——包数预算和时间预算：

// net/core/dev.c
static __latent_entropy void net_rx_action(struct softirq_action *h)
{
    struct softnet_data *sd = this_cpu_ptr(&softnet_data);
    unsigned long time_limit = jiffies +
        usecs_to_jiffies(READ_ONCE(net_hotdata.netdev_budget_usecs));
    int budget = READ_ONCE(net_hotdata.netdev_budget);
    LIST_HEAD(list);
    LIST_HEAD(repoll);

    sd->in_net_rx_action = true;
    local_irq_disable();
    list_splice_init(&sd->poll_list, &list);
    local_irq_enable();

    for (;;) {
        struct napi_struct *n;

        // 退出条件检查
        if (list_empty(&list)) {
            if (list_empty(&repoll)) break;
            list_splice_init(&repoll, &list);
        }

        n = list_first_entry(&list, struct napi_struct, poll_list);
        budget -= napi_poll(n, &repoll);

        // 预算耗尽或时间到？
        if (unlikely(budget <= 0 ||
                     time_after_eq(jiffies, time_limit))) {
            sd->time_squeeze++;
            break;
        }
    }

    // 处理未完成的 NAPI
    local_irq_disable();
    list_splice_tail_init(&sd->poll_list, &list);
    list_splice_tail(&repoll, &list);
    if (!list_empty(&list))
        __raise_softirq_irqoff(NET_RX_SOFTIRQ);
    list_splice(&list, &sd->poll_list);
    sd->in_net_rx_action = false;
    local_irq_restore(flags);
}

两个关键参数（可以通过 sysctl 调整）：

3.2 napi_poll：调用驱动的 poll 函数

napi_poll() 是对驱动 poll 函数的封装：

// net/core/dev.c (简化)
static int napi_poll(struct napi_struct *n, struct list_head *repoll)
{
    int work, weight;

    list_del_init(&n->poll_list);

    weight = n->weight;  // 通常是 64
    work = n->poll(n, weight);  // 调用驱动的 poll 回调

    if (work < weight) {
        // 包处理完了，退出轮询模式
        napi_complete_done(n, work);
    } else {
        // 还有更多包，重新加入轮询列表
        list_add_tail(&n->poll_list, repoll);
    }
    return work;
}

每个 NAPI 实例的 weight（默认 64）控制每次 poll 最多处理多少个包。这里有一个两级预算机制：

• NAPI 级：每次 napi->poll() 最多处理 weight（64）个包
• 全局级：每次 net_rx_action 最多处理 netdev_budget（300）个包

3.3 napi_complete_done：退出轮询、重开中断

当驱动 poll 函数处理的包数少于 weight，说明 ring buffer 已经清空，调用 napi_complete_done() 退出轮询模式：

// include/linux/netdevice.h:551
bool napi_complete_done(struct napi_struct *n, int work_done);

这个函数的核心操作：

1. 清除 NAPI_STATE_SCHED 标志——允许新的中断再次调度 NAPI
2. 如果 defer_hard_irqs > 0，使用 hrtimer 延迟重新开启中断（busy polling 优化）
3. 否则立即重新开启网卡中断

中断和轮询的切换逻辑是 NAPI 的精髓：

• 包少时：poll 很快完成 → napi_complete_done → 开中断 → 等下一个中断唤醒
• 包多时：poll 用完 weight → 不开中断 → net_rx_action 继续轮询 → 避免中断风暴

四、驱动 poll 函数：从 ring buffer 到 sk_buff

驱动的 poll 回调函数是连接硬件和内核网络栈的桥梁。以典型的 Intel 驱动为例：

// 驱动 poll 函数的典型伪代码
int driver_poll(struct napi_struct *napi, int budget)
{
    struct rx_ring *ring = container_of(napi, ...);
    int cleaned = 0;

    while (cleaned < budget) {
        struct rx_desc *desc = &ring->desc[ring->next_to_clean];

        // 检查描述符是否已被 DMA 填充
        if (!rx_desc_done(desc))
            break;

        // 分配或复用 sk_buff
        struct sk_buff *skb = build_skb_from_page(desc);

        // 填充 skb 元数据
        skb->protocol = eth_type_trans(skb, napi->dev);
        skb->ip_summed = CHECKSUM_UNNECESSARY;  // 硬件已校验

        // 交给 GRO
        napi_gro_receive(napi, skb);

        cleaned++;
        ring->next_to_clean = next_desc(ring);
    }

    // 补充 ring buffer
    driver_alloc_rx_buffers(ring);

    if (cleaned < budget)
        napi_complete_done(napi, cleaned);

    return cleaned;
}

4.1 eth_type_trans：L2 头解析

eth_type_trans() 是一个被广泛忽略但至关重要的函数。它在驱动 poll 中被调用，负责两件事：

// include/linux/etherdevice.h:40
__be16 eth_type_trans(struct sk_buff *skb, struct net_device *dev);

第一，设置 skb->protocol——从以太网帧头的 EtherType 字段提取协议类型。这决定了后面 __netif_receive_skb_core 把包分发给哪个 L3 协议处理函数。常见值：

• ETH_P_IP（0x0800）→ 交给 ip_rcv()
• ETH_P_IPV6（0x86DD）→ 交给 ipv6_rcv()
• ETH_P_ARP（0x0806）→ 交给 arp_rcv()
• ETH_P_8021Q（0x8100）→ VLAN 处理

第二，设置 skb->pkt_type——判断这个帧是发给本机的（PACKET_HOST）、广播的（PACKET_BROADCAST）、多播的（PACKET_MULTICAST）、还是发给别人被混杂模式抓到的（PACKET_OTHERHOST）。判断依据是帧的目标 MAC 地址和 dev->dev_addr 的比较。

第三，调整 skb->data 指针——通过 skb_pull_inline(skb, ETH_HLEN) 跳过以太网帧头（14 字节），让 skb->data 指向 IP 头。这就是 第一篇 里讲的”剥洋葱”操作。

4.2 硬件卸载标志

现代网卡在 DMA 描述符中附带校验和验证结果。驱动据此设置 skb->ip_summed：

CHECKSUM_UNNECESSARY 在高性能场景中至关重要——跳过软件校验可以节省显著的 CPU。

五、GRO：在交给协议栈之前先聚合

GRO（Generic Receive Offload）在 napi_gro_receive() 中执行，目的是把多个属于同一 TCP 流的小包聚合成一个大包再交给协议栈——减少协议栈处理次数，降低 per-packet 开销。

5.1 napi_gro_receive 的工作流程

// include/linux/netdevice.h:3997
gro_result_t napi_gro_receive(struct napi_struct *napi, struct sk_buff *skb);

napi_gro_receive() 的内部逻辑：

1. 查找匹配的 GRO 聚合链：在 napi->gro_hash[]（8 个桶，按流哈希索引）中查找可以聚合的已有 skb
2. 调用协议的 GRO 回调：通过 packet_offload.gro_receive() 让各协议层判断能否聚合
3. 三种结果：
   • GRO_MERGED：成功聚合到已有 skb，当前 skb 被释放
   • GRO_HELD：当前 skb 被加入 GRO 链表，等待后续包来聚合
   • GRO_NORMAL：无法聚合，直接交给 netif_receive_skb()

5.2 GRO 的聚合条件

GRO 聚合需要满足多个条件（以 TCP 为例）：

• 同一个五元组（源/目标 IP、源/目标端口、协议）
• TCP 序列号连续
• TCP flags 相同（不能一个有 PSH 一个没有——实际上 GRO 会保留最后一个包的 PSH 标志）
• IP TTL、TOS 相同
• 没有 IP options
• 聚合后的包大小不超过 GRO_MAX_SIZE（默认 65536 字节）

5.3 GRO flush 时机

聚合不能无限等待。以下情况会触发 GRO flush，把聚合好的大包交给协议栈：

• napi_gro_flush()：驱动 poll 结束时（napi_complete_done 内部调用）
• 超时：skb 在 GRO 链表中待的时间超过 gro_flush_timeout（可通过 sysctl 设置）
• 聚合数量上限：MAX_GRO_SKBS（默认 8 个包）
• 协议层拒绝：收到乱序包、FIN、RST 等

5.4 GRO_NORMAL 路径

在 6.x 内核中，GRO 完成的 skb 不是直接调用 netif_receive_skb()，而是被加入 napi->rx_list，达到 gro_normal_batch（默认 8）个后批量交给 netif_receive_skb_list()——这进一步减少了函数调用开销。

// include/linux/netdevice.h:375-376
struct list_head    rx_list;    // Pending GRO_NORMAL skbs
int                 rx_count;   // length of rx_list

六、netif_receive_skb 与协议分发

netif_receive_skb() 是整个收包路径中L2 → L3 的分水岭。

// include/linux/netdevice.h:3993
int netif_receive_skb(struct sk_buff *skb);

6.1 RPS 分发

如果启用了 RPS（Receive Packet Steering），netif_receive_skb() 首先会根据包的流哈希选择一个目标 CPU，通过 IPI（Inter-Processor Interrupt）把 skb 转移到目标 CPU 的 input_pkt_queue，然后由目标 CPU 的 process_backlog() 继续处理。

RPS 的好处是把协议栈处理分散到多个 CPU，缺点是增加了 IPI 的延迟和缓存失效。在有硬件 RSS 的场景下，RPS 通常不需要开启。

如果没有启用 RPS（或者 RPS 选择了当前 CPU），直接进入 __netif_receive_skb() → __netif_receive_skb_core()。

6.2 __netif_receive_skb_core：核心分发逻辑

这个函数是协议分发的核心——它决定每个 skb 交给谁处理。

// net/core/dev.c (简化关键路径)
static int __netif_receive_skb_core(struct sk_buff **pskb, bool pfmemalloc,
                                    struct packet_type **ppt_prev)
{
    struct sk_buff *skb = *pskb;
    struct net_device *orig_dev = skb->dev;
    struct packet_type *ptype, *pt_prev = NULL;
    int ret;

    // 1. 交给 ptype_all 链表上的所有注册者（如 tcpdump / AF_PACKET）
    list_for_each_entry_rcu(ptype, &net_hotdata.ptype_all, list) {
        if (pt_prev)
            ret = deliver_skb(skb, pt_prev, orig_dev);
        pt_prev = ptype;
    }

    // 2. 处理 VLAN tag
    skb_vlan_untag(skb);

    // 3. 调用 rx_handler（bridge / OVS / macvlan 等虚拟设备的钩子）
    rx_handler = rcu_dereference(skb->dev->rx_handler);
    if (rx_handler) {
        switch (rx_handler(&skb)) {
        case RX_HANDLER_CONSUMED: return NET_RX_SUCCESS;
        case RX_HANDLER_ANOTHER: goto another_round;  // 重新分发
        case RX_HANDLER_EXACT:   deliver_exact = true; break;
        case RX_HANDLER_PASS:    break;
        }
    }

    // 4. 根据 skb->protocol 在 ptype_base[] 哈希表中查找协议处理函数
    type = skb->protocol;
    deliver_ptype_list_skb(skb, &pt_prev, orig_dev, type,
                           &ptype_base[ntohs(type) & PTYPE_HASH_MASK]);

    // 5. 调用协议处理函数（如 ip_rcv）
    if (pt_prev) {
        ret = pt_prev->func(skb, skb->dev, pt_prev, orig_dev);
    }
    return ret;
}

6.3 三条分发链

这段代码揭示了三条独立的分发链：

第一条：ptype_all——全协议抓包

ptype_all 是一个链表，所有用 AF_PACKET socket 注册的抓包程序都挂在这里。典型的用户是 tcpdump。每个经过 __netif_receive_skb_core 的 skb 都会被遍历一次 ptype_all——这就是为什么在高 PPS 场景下运行 tcpdump 会显著增加 CPU 开销。

// 注册方式（AF_PACKET socket 创建时）
struct packet_type pt;
pt.type = htons(ETH_P_ALL);
pt.func = packet_rcv;   // AF_PACKET 的接收函数
dev_add_pack(&pt);       // 加入 ptype_all

第二条：rx_handler——虚拟设备钩子

每个 net_device 可以注册一个 rx_handler。这是 bridge、OVS、macvlan 等虚拟网络设备截获收包路径的入口。例如，当一个网卡被加入 Linux bridge 时，bridge 模块会注册 br_handle_frame 作为 rx_handler——所有从这个网卡收到的包都先经过 bridge 转发逻辑。

rx_handler 的四种返回值控制后续行为：

第三条：ptype_base[hash]——协议类型分发

ptype_base[] 是一个以 skb->protocol 的哈希值索引的链表数组。IP 协议在初始化时注册：

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

// 初始化时
dev_add_pack(&ip_packet_type);

这样，所有 skb->protocol == ETH_P_IP 的包都会调用 ip_rcv() 进入 IP 层处理。

七、IP 层：ip_rcv → 路由决策 → ip_local_deliver

7.1 ip_rcv：入口校验

// include/net/ip.h:161
int ip_rcv(struct sk_buff *skb, struct net_device *dev,
           struct packet_type *pt, struct net_device *orig_dev);

ip_rcv() 的主要工作是校验 IP 头的合法性：

1. 检查 skb->pkt_type——如果是 PACKET_OTHERHOST（不是发给本机的），直接丢弃
2. 如果 skb 被共享（skb_shared(skb) 为真），先 skb_clone() 复制一份
3. 拉取 IP 头——确保 skb->data 覆盖完整的 IP 头（至少 20 字节）
4. 校验 IP 头的基本字段：版本号（必须是 4）、头长度（≥ 5 即 20 字节）、总长度
5. 校验 IP 头 checksum（如果硬件没有验证过）

校验通过后，调用 Netfilter 的 NF_INET_PRE_ROUTING 钩子——这是 iptables PREROUTING 链、conntrack 入口、DNAT 执行点：

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

7.2 ip_rcv_finish：路由查找

Netfilter PREROUTING 通过后，进入 ip_rcv_finish()。核心操作是路由查找：

// net/ipv4/ip_input.c (简化)
static int ip_rcv_finish(struct net *net, struct sock *sk, struct sk_buff *skb)
{
    struct iphdr *iph = ip_hdr(skb);

    // 如果还没有路由缓存（dst_entry），执行路由查找
    if (!skb_valid_dst(skb)) {
        int err = ip_route_input_noref(skb, iph->daddr, iph->saddr,
                                       iph->tos, skb->dev);
        if (err)
            goto drop;
    }

    // 根据路由结果调用相应的处理函数
    return dst_input(skb);
}

ip_route_input_noref() 查询 FIB（Forwarding Information Base）路由表，结果挂在 skb->_skb_refdst 上。路由查找的结果决定了包的去向：

7.3 ip_local_deliver：上送传输层

// include/net/ip.h:165
int ip_local_deliver(struct sk_buff *skb);

对于发给本机的包，ip_local_deliver() 做两件事：

1. IP 分片重组：如果 IP 头的 MF（More Fragments）标志位或 Fragment Offset 不为零，调用 ip_defrag() 进行分片重组。重组完成前，包暂存在分片队列中
2. NF_INET_LOCAL_IN 钩子：调用 Netfilter 的 INPUT 链——这是 iptables INPUT 规则的执行点

int ip_local_deliver(struct sk_buff *skb)
{
    struct net *net = dev_net(skb->dev);

    if (ip_is_fragment(ip_hdr(skb))) {
        if (ip_defrag(net, skb, IP_DEFRAG_LOCAL_DELIVER))
            return 0;  // 分片未完成，等待更多分片
    }

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

7.4 ip_local_deliver_finish → ip_protocol_deliver_rcu

Netfilter INPUT 通过后，ip_local_deliver_finish() 剥掉 IP 头，根据 IP 头中的 protocol 字段（TCP=6, UDP=17, ICMP=1）调用注册在 inet_protos[] 数组中的传输层处理函数：

// include/net/ip.h:166
void ip_protocol_deliver_rcu(struct net *net, struct sk_buff *skb, int proto);

// net/ipv4/ip_input.c (简化)
void ip_protocol_deliver_rcu(struct net *net, struct sk_buff *skb, int protocol)
{
    const struct net_protocol *ipprot;

    ipprot = rcu_dereference(inet_protos[protocol]);
    if (ipprot) {
        ret = INDIRECT_CALL_2(ipprot->handler, tcp_v4_rcv, udp_rcv, skb);
    } else {
        // 没有注册处理函数，发送 ICMP Destination Unreachable
        icmp_send(skb, ICMP_DEST_UNREACH, ICMP_PROT_UNREACH, 0);
    }
}

注意这里使用了 INDIRECT_CALL_2 宏——它通过编译时的直接调用分支提示帮助 CPU 分支预测器，减少 indirect call 的 retpoline 开销。这在 Spectre 缓解措施启用后尤为重要。

八、传输层：tcp_v4_rcv 与 socket 锁竞争

8.1 tcp_v4_rcv 入口

// include/net/tcp.h:335
int tcp_v4_rcv(struct sk_buff *skb);

tcp_v4_rcv() 是 TCP 收包的入口。它的主要工作流程：

// net/ipv4/tcp_ipv4.c (简化关键路径)
int tcp_v4_rcv(struct sk_buff *skb)
{
    const struct tcphdr *th;
    struct sock *sk;

    // 1. 拉取并校验 TCP 头
    th = (const struct tcphdr *)skb->data;
    if (skb->len < sizeof(struct tcphdr))
        goto bad_packet;

    // 2. 查找匹配的 socket
    sk = __inet_lookup_skb(net, net->ipv4.tcp_death_row.hashinfo,
                           skb, __tcp_hdrlen(th),
                           th->source, th->dest,
                           iph->saddr, iph->daddr,
                           inet_iif(skb), sdif);
    if (!sk)
        goto no_tcp_socket;

    // 3. 时间戳处理 (PAWS)
    if (!tcp_filter(sk, skb)) {
        th = (const struct tcphdr *)skb->data;
        iph = ip_hdr(skb);
        tcp_v4_fill_cb(skb, iph, th);
    }

    // 4. 根据 socket 锁状态决定处理方式
    if (sk->sk_state == TCP_LISTEN) {
        // 监听 socket：处理 SYN / ACK-of-SYN
        ret = tcp_v4_do_rcv(sk, skb);
    } else {
        bh_lock_sock_nested(sk);
        if (!sock_owned_by_user(sk)) {
            // socket 未被用户态锁住，直接处理
            ret = tcp_v4_do_rcv(sk, skb);
        } else {
            // socket 被用户态锁住（正在 sendmsg/recvmsg 等）
            // 把包加入 backlog 队列
            if (!tcp_add_backlog(sk, skb))
                goto discard_and_relse;
        }
        bh_unlock_sock(sk);
    }
    return ret;
}

8.2 Early Demux 优化

在 ip_rcv_finish 阶段（甚至更早），如果启用了 Early Demux（默认开启），内核会尝试提前查找 socket：

// include/net/tcp.h:334
int tcp_v4_early_demux(struct sk_buff *skb);

Early Demux 在路由查找之前就通过四元组找到 socket，直接复用 socket 上缓存的路由信息（sk->sk_rx_dst），从而跳过完整的 FIB 路由查找。对于已建立连接的 TCP 流量，这可以显著减少路由查找开销。

可以通过 sysctl 控制：

net.ipv4.tcp_early_demux = 1    # 默认开启
net.ipv4.udp_early_demux = 1    # UDP 也支持

8.3 socket 锁竞争与 backlog 队列

这是收包路径中最容易被忽视的性能关键点。

当 tcp_v4_rcv() 在软中断上下文中被调用时，它需要访问 socket。但如果此时用户态正在对同一个 socket 执行 sendmsg() 或 recvmsg()（持有 sk->sk_lock.owned），软中断不能直接处理这个包——因为修改 socket 状态需要持锁。

解决方案就是 backlog 队列：

// include/net/sock.h:399
struct {
    atomic_t    rmem_alloc;     // 接收内存计数
    int         len;            // backlog 队列总长度（字节）
    struct sk_buff  *head;      // 队列头
    struct sk_buff  *tail;      // 队列尾
} sk_backlog;

当 socket 被锁住时，sk_add_backlog() 把 skb 暂存到 backlog 队列：

// include/net/sock.h:1065
static inline __must_check int sk_add_backlog(struct sock *sk,
                                              struct sk_buff *skb,
                                              unsigned int limit)
{
    if (sk_rcvqueues_full(sk, limit))
        return -ENOBUFS;    // 队列满，丢包

    __sk_add_backlog(sk, skb);
    sk->sk_backlog.len += skb->truesize;
    return 0;
}

backlog 的处理时机：当用户态释放 socket 锁时（release_sock() 内部），会调用 __release_sock() 遍历 backlog 队列，对每个 skb 调用 sk->sk_backlog_rcv()（对 TCP 来说就是 tcp_v4_do_rcv()）。

// include/net/sock.h:1089
static inline int sk_backlog_rcv(struct sock *sk, struct sk_buff *skb)
{
    return INDIRECT_CALL_INET(sk->sk_backlog_rcv,
                              tcp_v6_do_rcv,
                              tcp_v4_do_rcv,
                              sk, skb);
}

性能影响：在高并发场景下，如果用户态频繁持锁（大量小包的 recvmsg()），backlog 队列会成为瓶颈——软中断只能把包堆在 backlog 里，等用户态释放锁后才能处理。这会增加包的处理延迟，也会占用更多内存。

九、Socket 接收队列：从内核到用户态

9.1 两级队列模型

TCP socket 的收包实际上涉及两个队列：

当 tcp_v4_do_rcv() 发现包是有序数据（序列号匹配 tp->rcv_nxt），它通过快速路径直接把数据加入 sk_receive_queue。如果是乱序数据，则加入 tp->out_of_order_queue 等待后续包到齐后重组。

9.2 sock_queue_rcv_skb：通用入队函数

// include/net/sock.h:2469
static inline int sock_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
{
    return sock_queue_rcv_skb_reason(sk, skb, NULL);
}

// include/net/sock.h:2466
int sock_queue_rcv_skb_reason(struct sock *sk, struct sk_buff *skb,
                              enum skb_drop_reason *reason);

这是 UDP 和其他协议常用的入队函数（TCP 有自己的快速路径，不一定经过这里）。它的核心逻辑：

1. 检查 socket 接收缓冲区是否已满（sk->sk_rmem_alloc vs sk->sk_rcvbuf）
2. 调用 socket filter（如果有 BPF filter 附加在 socket 上）
3. skb_set_owner_r(skb, sk) —— 设置 skb 的 owner，并增加 sk_rmem_alloc
4. __skb_queue_tail(&sk->sk_receive_queue, skb) —— 加入接收队列尾部

9.3 sk_data_ready：唤醒等待的进程

入队完成后，调用 sk->sk_data_ready() 通知等待在这个 socket 上的进程。默认实现是 sock_def_readable()：

void sock_def_readable(struct sock *sk)
{
    struct socket_wq *wq;

    rcu_read_lock();
    wq = rcu_dereference(sk->sk_wq);
    if (skwq_has_sleeper(wq))
        wake_up_interruptible_sync_poll(&wq->wait,
                                         EPOLLIN | EPOLLRDNORM);
    sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
    rcu_read_unlock();
}

如果有进程在 recvmsg() 中阻塞等待（或者在 epoll_wait() 中等待这个 fd），wake_up_interruptible_sync_poll() 会唤醒它。

9.4 用户态读取路径

被唤醒的进程调用 recvmsg() → tcp_recvmsg() / udp_recvmsg()，从 sk_receive_queue 中取出 skb，通过 skb_copy_datagram_msg() 把数据从内核缓冲区拷贝到用户态缓冲区（copy_to_user()）。

这是标准收包路径中唯一一次真正的数据拷贝——从内核内存到用户态内存。之前的所有阶段都只是在操作 sk_buff 的指针和元数据。

十、可观测性：用 bpftrace 追踪收包全路径

10.1 追踪各阶段延迟分布

以下 bpftrace 脚本测量一个包从 napi_gro_receive 到 tcp_v4_rcv 的延迟：

#!/usr/bin/env bpftrace

// 记录 GRO 收包时间戳
kprobe:napi_gro_receive
{
    $skb = (struct sk_buff *)arg1;
    @gro_ts[arg1] = nsecs;
}

// 在 tcp_v4_rcv 入口测量延迟
kprobe:tcp_v4_rcv
{
    $skb = (struct sk_buff *)arg0;
    $ts = @gro_ts[(uint64)$skb];
    if ($ts > 0) {
        @latency_us = hist((nsecs - $ts) / 1000);
        delete(@gro_ts[(uint64)$skb]);
    }
}

END { clear(@gro_ts); }

10.2 监控 softirq 被挤压

#!/usr/bin/env bpftrace

// 监控 time_squeeze：net_rx_action 因预算耗尽退出的次数
tracepoint:net:netif_receive_skb
{
    @rx_cpu[cpu] = count();
}

// 每秒输出 per-CPU 收包统计
interval:s:1
{
    print(@rx_cpu);
    clear(@rx_cpu);
}

也可以直接查看 /proc/net/softnet_stat：

# 每行对应一个 CPU，三列分别是：processed, dropped, time_squeeze
cat /proc/net/softnet_stat

10.3 追踪 backlog 队列深度

#!/usr/bin/env bpftrace

// 在 sk_add_backlog 时追踪队列深度
kprobe:sk_add_backlog
{
    $sk = (struct sock *)arg0;
    $skb = (struct sk_buff *)arg1;
    @backlog_len = hist($sk->sk_backlog.len);
}

interval:s:5
{
    print(@backlog_len);
    clear(@backlog_len);
}

10.4 追踪丢包点

内核在丢弃 skb 时统一调用 kfree_skb_reason()（6.x 内核），附带丢弃原因枚举：

#!/usr/bin/env bpftrace

tracepoint:skb:kfree_skb
{
    @drop_reason[args->reason] = count();
    @drop_location[ksym(args->location)] = count();
}

interval:s:10
{
    print(@drop_reason);
    print(@drop_location);
    clear(@drop_reason);
    clear(@drop_location);
}

常见丢包位置和原因：

10.5 per-function CPU 热点

# 收包相关函数的 CPU 热点采样
perf top -g -e cycles:k --filter='net_rx_action or \
    __netif_receive_skb or ip_rcv or tcp_v4_rcv'

# 生成火焰图
perf record -a -g -e cycles:k -- sleep 10
perf script | stackcollapse-perf.pl | flamegraph.pl > rx-flame.svg

十一、关键性能参数与调优

调优示例：高 PPS 场景

# 增大软中断预算
sysctl -w net.core.netdev_budget=600
sysctl -w net.core.netdev_budget_usecs=4000

# 开启 GRO 延迟 flush（减少协议栈入口次数）
echo 20000 > /sys/class/net/eth0/gro_flush_timeout
echo 1 > /sys/class/net/eth0/napi_defer_hard_irqs

# 增大 ring buffer
ethtool -G eth0 rx 4096

# 检查 time_squeeze 是否缓解
watch -n 1 'cat /proc/net/softnet_stat'

参考文献

1. Linux 内核源码，net/core/dev.c，6.6 LTS / 6.8
2. Linux 内核源码，net/ipv4/ip_input.c，6.6 LTS
3. Linux 内核源码，net/ipv4/tcp_ipv4.c，6.6 LTS
4. Linux 内核源码，include/linux/netdevice.h，6.8（napi_struct 定义于第 352 行，softnet_data 定义于第 3281 行）
5. Linux 内核源码，include/net/sock.h，6.8（sk_backlog 定义于第 399 行，sk_add_backlog 定义于第 1065 行）
6. Linux 内核源码，include/linux/etherdevice.h，6.8（eth_type_trans 声明于第 40 行）
7. Linux 内核文档，Documentation/networking/napi.rst
8. Linux 内核文档，Documentation/networking/scaling.rst（RSS/RPS/RFS/XPS）

上一篇：net_device 与网卡驱动模型：从硬件到内核的接口契约

下一篇：发包路径全解：从 send() 到网线