Linux 网络内核协议栈深度剖析（二）：网卡驱动与 NAPI 收发包机制

Posted on In ,

本文是 Linux 网络内核协议栈系列的第二篇，聚焦于网卡驱动框架与 NAPI（New API）收发包机制。我们将深入 Linux 6.4-rc1 源码，逐层拆解从网卡硬件中断到 sk_buff 进入协议栈的完整链路，并对称地分析发送路径。所有代码片段均直接取自内核源文件，确保与实际内核行为一致。

在开始之前，有必要建立一个整体性的认知框架：Linux 网络子系统的收发包路径，本质上是一套生产者-消费者模型。在接收方向，网卡硬件是生产者，不断将数据帧写入 DMA ring buffer；NAPI softirq 是消费者，以 batch 的方式将这些帧送入协议栈。在发送方向，上层协议（TCP/UDP）是生产者，将 sk_buff 投入 qdisc 队列；驱动 ndo_start_xmit 是最终消费者，将帧写入硬件 TX ring 触发 DMA。

理解这条路径上每个关键函数的边界和契约，是进行高性能网络编程、内核旁路（kernel bypass）优化以及网络问题根因分析的必要前提。

一、网卡驱动框架：struct net_device 与 net_device_ops

1.1 网络设备的核心抽象

Linux 内核中，每一张网卡——无论是物理以太网卡、虚拟 tun/tap 设备还是 loopback——在内核里都对应一个 struct net_device 实例。这个结构体定义于 include/linux/netdevice.h（第 2062 行），是驱动层与网络核心层（net/core/）之间最重要的接口契约。

从内存布局的角度来看，net_device 被精心划分为若干 cache line，以减少多核场景下的伪共享：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
// include/linux/netdevice.h: 2062
struct net_device {
    char            name[IFNAMSIZ];
    ...
    unsigned long   state;

    struct list_head    dev_list;
    struct list_head    napi_list;    /* 挂载在此设备上的所有 napi_struct */
    struct list_head    unreg_list;
    struct list_head    close_list;
    struct list_head    ptype_all;
    struct list_head    ptype_specific;
    ...
    /* 快速路径 cache line */
    unsigned int        flags;
    const struct net_device_ops *netdev_ops;   /* 驱动操作函数集 */
    int                 ifindex;
    unsigned int        mtu;

    netdev_features_t   features;      /* 已启用的 offload 特性 */
    netdev_features_t   hw_features;   /* 硬件支持但未必启用 */
    ...
    const struct ethtool_ops *ethtool_ops;

    /* 接收路径 cache line */
    struct netdev_rx_queue  *_rx;
    unsigned int            num_rx_queues;
    unsigned int            real_num_rx_queues;
    unsigned long           gro_flush_timeout;
    int                     napi_defer_hard_irqs;

    /* 发送路径 cache line */
    struct netdev_queue *_tx ____cacheline_aligned_in_smp;
    unsigned int         num_tx_queues;
    unsigned int         real_num_tx_queues;
    struct Qdisc __rcu  *qdisc;
    unsigned int         tx_queue_len;
};

几个关键字段值得专门说明：

  • **napi_list**：链表头，挂载了所有注册到该设备的 napi_struct。一张多队列网卡通常为每个 RX 队列注册一个 NAPI 实例，它们全部挂在此链表上。
  • features：netdev_features_t 类型的位图（定义于 include/linux/netdev_features.h），记录当前已激活的 offload 能力。常见标志包括：
    • NETIF_F_TSO（TCP Segmentation Offload，TCPv4 分段卸载）
    • NETIF_F_TSO6（TCPv6 分段卸载）
    • NETIF_F_GRO（Generic Receive Offload，通用接收聚合）
    • NETIF_F_RXCSUM（接收校验和卸载）
    • NETIF_F_HW_CSUM（硬件计算所有协议校验和）
    • NETIF_F_SG（Scatter/Gather I/O，分散聚集 DMA）
  • **num_tx_queues / num_rx_queues**：在 alloc_netdev_mq() 时分配的队列数量；real_num_* 是当前实际激活的数量，驱动可在运行时调整。
  • **tx_queue_len**：对应 ip link 命令看到的 txqueuelen，是默认 qdisc 的软件队列深度（通常 1000）。
  • gro_flush_timeout 与 **napi_defer_hard_irqs**：GRO 延迟 flush 相关的调优参数，通过 /sys/class/net/<dev>/gro_flush_timeout 和 napi_defer_hard_irqs 配置，适用于高吞吐低延迟的折中场景。
  • **qdisc**：指向该设备当前根 qdisc（流量控制队列），默认为 pfifo_fast（FIFO 三优先级队列）。在 tc qdisc replace 之后此指针会被 RCU 方式更新。

值得注意的是 net_device 并非驱动直接分配，而是通过 alloc_netdev_mq(priv_size, name, setup_fn, num_queues) 分配，私有数据紧跟 struct net_device 之后，通过 netdev_priv(dev) 访问。register_netdev() 将设备注册进全局 net_namespace 并触发 NETDEV_REGISTER 通知链，完成设备的可见化。

1.2 struct net_device_ops：驱动操作函数集

驱动通过填充 struct net_device_ops（include/linux/netdevice.h，第 1419 行）并赋值给 dev->netdev_ops 来向内核注册其能力：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// include/linux/netdevice.h: 1419
struct net_device_ops {
    int         (*ndo_init)(struct net_device *dev);
    void        (*ndo_uninit)(struct net_device *dev);
    int         (*ndo_open)(struct net_device *dev);
    int         (*ndo_stop)(struct net_device *dev);
    netdev_tx_t (*ndo_start_xmit)(struct sk_buff *skb,
                                  struct net_device *dev);
    netdev_features_t (*ndo_features_check)(struct sk_buff *skb,
                                            struct net_device *dev,
                                            netdev_features_t features);
    u16         (*ndo_select_queue)(struct net_device *dev,
                                   struct sk_buff *skb,
                                   struct net_device *sb_dev);
    ...
    void        (*ndo_get_stats64)(struct net_device *dev,
                                   struct rtnl_link_stats64 *storage);
    ...
    int         (*ndo_change_mtu)(struct net_device *dev, int new_mtu);
    void        (*ndo_tx_timeout)(struct net_device *dev,
                                  unsigned int txqueue);
};

核心回调的语义：

回调 触发时机 典型操作
ndo_open ip link set eth0 up 分配 RX/TX ring buffer、注册中断、启动 NAPI
ndo_stop ip link set eth0 down 停止 DMA、注销中断、释放 ring buffer
ndo_start_xmit 内核发送路径最终调用 将 sk_buff 写入 TX descriptor ring，触发 DMA
ndo_get_stats64 ip -s link 或 ethtool -S 从驱动 per-CPU 计数器中读取统计数据
ndo_select_queue 多队列发送，选择 TX 队列 驱动可基于 skb 元数据做定制化映射

ethtool_ops 则是另一套接口，专门服务于 ethtool 工具对硬件参数的查询和配置（link speed、autoneg、ring size、coalesce 参数等），与数据平面的 net_device_ops 完全解耦。

驱动编写的核心工作就是按照这两套接口填充函数指针，然后通过 alloc_etherdev_mqs() + register_netdev() 将设备注册给内核。内核其余部分完全通过这两套接口与驱动交互，做到了驱动与协议栈的彻底分离。

二、中断与 NAPI 机制

2.1 传统中断模式的困境

在 NAPI 出现之前，网卡采用纯中断驱动模式：每收到一个数据帧，就触发一次硬件中断，内核在中断处理函数（ISR）中分配 sk_buff、复制数据、入队，然后返回。

这种模式在低负载时工作良好，但在高流量场景（如 10GbE/100GbE）下会引发中断风暴（interrupt storm）：每秒数百万次中断会耗尽 CPU 资源，大量时间消耗在中断上下文切换和 ISR 本身，实际吞吐反而下降。测量发现，当包速率超过约 10 万 pps 时，传统中断模式的 CPU 利用率就已饱和。

NAPI 的解决思路是中断 + 轮询混合模式：收到第一个包时触发中断，在中断处理函数中关闭硬件中断并调度 softirq 轮询；softirq 以 budget 为上限，批量处理队列中已到达的帧；处理完毕或 budget 耗尽后才重新开启中断。这将中断次数从 O(包数) 降低到 O(批次数)。

2.2 struct napi_struct：NAPI 实例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
// include/linux/netdevice.h: 348
struct napi_struct {
    /* poll_list 仅由持有 NAPI_STATE_SCHED bit 的实体管理 */
    struct list_head    poll_list;

    unsigned long       state;
    int                 weight;              /* 每次轮询最多处理的包数（budget） */
    int                 defer_hard_irqs_count;
    unsigned long       gro_bitmask;         /* 记录哪些 GRO hash bucket 非空 */
    int                 (*poll)(struct napi_struct *, int); /* 驱动注册的轮询函数 */
#ifdef CONFIG_NETPOLL
    int                 poll_owner;
#endif
    int                 list_owner;          /* 当前在哪个 CPU 的 poll_list 上 */
    struct net_device  *dev;
    struct gro_list     gro_hash[GRO_HASH_BUCKETS]; /* GRO 合包哈希表 */
    struct sk_buff     *skb;
    struct list_head    rx_list;             /* 待处理的 GRO_NORMAL skbs */
    int                 rx_count;
    unsigned int        napi_id;
    struct hrtimer      timer;               /* GRO flush 定时器 */
    struct task_struct *thread;              /* threaded NAPI 内核线程 */
    /* control path 字段 */
    struct list_head    dev_list;
    struct hlist_node   napi_hash_node;
};

state 字段的位掩码由 NAPIF_STATE_* 系列常量定义，核心状态包括：

  • NAPI_STATE_SCHED：NAPI 已被调度，正在或即将进行 poll
  • NAPI_STATE_MISSED：poll 期间再次触发了调度，需要重新 poll
  • NAPI_STATE_DISABLE：NAPI 正在被禁用（napi_disable() 中）
  • NAPI_STATE_THREADED：使用独立内核线程而非 softirq 来执行 poll

2.3 ____napi_schedule：将 NAPI 加入 softirq 轮询列表

当网卡 ISR 检测到新数据时，调用链是 napi_schedule() → napi_schedule_prep() → __napi_schedule() → ____napi_schedule()：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// net/core/dev.c: 4332
static inline void ____napi_schedule(struct softnet_data *sd,
                                     struct napi_struct *napi)
{
    struct task_struct *thread;

    lockdep_assert_irqs_disabled();

    if (test_bit(NAPI_STATE_THREADED, &napi->state)) {
        thread = READ_ONCE(napi->thread);
        if (thread) {
            if (READ_ONCE(thread->__state) != TASK_INTERRUPTIBLE)
                set_bit(NAPI_STATE_SCHED_THREADED, &napi->state);
            wake_up_process(thread);
            return;
        }
    }

    list_add_tail(&napi->poll_list, &sd->poll_list);  /* 加入本 CPU 的轮询列表 */
    WRITE_ONCE(napi->list_owner, smp_processor_id());
    /* 如果不在 net_rx_action() 上下文中，需要主动 raise softirq */
    if (!sd->in_net_rx_action)
        __raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

softnet_data 是 per-CPU 结构，其中的 poll_list 是待轮询的 NAPI 实例链表。__napi_schedule() 的实现为：

1
2
3
4
5
6
7
8
9
10
// net/core/dev.c: 5970
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);
}
EXPORT_SYMBOL(__napi_schedule);

napi_schedule_prep() 则通过 try_cmpxchg 原子地设置 NAPIF_STATE_SCHED 位，并处理重入情况——若已经处于 SCHED 状态，则设置 NAPIF_STATE_MISSED，保证不会遗漏数据。

2.4 net_rx_action：softirq 的核心调度器

NET_RX_SOFTIRQ 的处理函数是 net_rx_action()，在 softirq 上下文中执行（ksoftirqd 内核线程或中断返回路径）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
// net/core/dev.c: 6659
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(netdev_budget_usecs));
    int budget = READ_ONCE(netdev_budget);   /* 默认值 300 */
    LIST_HEAD(list);
    LIST_HEAD(repoll);

start:
    sd->in_net_rx_action = true;
    local_irq_disable();
    list_splice_init(&sd->poll_list, &list);  /* 原子地取走当前轮询列表 */
    local_irq_enable();

    for (;;) {
        struct napi_struct *n;

        skb_defer_free_flush(sd);

        if (list_empty(&list)) {
            if (list_empty(&repoll)) {
                sd->in_net_rx_action = false;
                barrier();
                if (!list_empty(&sd->poll_list))
                    goto start;
                if (!sd_has_rps_ipi_waiting(sd))
                    goto end;
            }
            break;
        }

        n = list_first_entry(&list, struct napi_struct, poll_list);
        budget -= napi_poll(n, &repoll);   /* 调用驱动 poll，消耗 budget */

        /* 超时或 budget 耗尽则退出，避免 softirq 饥饿其他任务 */
        if (unlikely(budget <= 0 ||
                     time_after_eq(jiffies, time_limit))) {
            sd->time_squeeze++;
            break;
        }
    }
    ...
}

netdev_budget 的默认值为 300，netdev_budget_usecs 默认为 2 * USEC_PER_SEC / HZ（在 HZ=1000 时为 2ms）。这两个参数均可通过 /proc/sys/net/core/ 调整。time_squeeze 计数器记录 budget 耗尽的次数，可在 /proc/net/softnet_stat 中观察，是判断 softirq 是否成为瓶颈的重要指标。

2.5 napi_poll：驱动 poll 函数的调用者

napi_poll() 负责从 poll_list 取出 NAPI 实例并调用其 poll 回调（即驱动注册的函数，如 igb 驱动的 igb_poll()）：

1
2
3
4
5
6
7
8
9
10
// net/core/dev.c: 6187 (简化)
static int napi_poll(struct napi_struct *n, struct list_head *repoll)
{
    int (*napi_poll)(struct napi_struct *napi, int budget);
    ...
    napi_poll = napi->poll;
    work = napi_poll(napi, budget);   /* 调用驱动注册的 poll 函数 */
    trace_napi_poll(napi, work, budget);
    ...
}

驱动的 poll 函数以 budget 为上限，从硬件 RX descriptor ring 中取帧，为每帧调用 napi_gro_receive() 或直接 netif_receive_skb()，最终返回实际处理的包数 work。若 work < budget，说明队列已清空，驱动调用 napi_complete_done() 重新开启硬件中断。

2.6 napi_complete_done：完成轮询，重新使能中断

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
// net/core/dev.c: 6031
bool napi_complete_done(struct napi_struct *n, int work_done)
{
    unsigned long flags, val, new, timeout = 0;
    bool ret = true;

    if (unlikely(n->state & (NAPIF_STATE_NPSVC |
                             NAPIF_STATE_IN_BUSY_POLL)))
        return false;

    if (work_done) {
        if (n->gro_bitmask)
            timeout = READ_ONCE(n->dev->gro_flush_timeout);
        n->defer_hard_irqs_count = READ_ONCE(n->dev->napi_defer_hard_irqs);
    }
    if (n->defer_hard_irqs_count > 0) {
        n->defer_hard_irqs_count--;
        timeout = READ_ONCE(n->dev->gro_flush_timeout);
        if (timeout)
            ret = false;   /* 继续用 hrtimer 驱动，推迟开中断 */
    }
    if (n->gro_bitmask)
        napi_gro_flush(n, !!timeout);  /* 刷出 GRO 聚合队列 */

    gro_normal_list(n);

    /* 从 poll_list 中移除自己 */
    if (unlikely(!list_empty(&n->poll_list))) {
        local_irq_save(flags);
        list_del_init(&n->poll_list);
        local_irq_restore(flags);
    }
    WRITE_ONCE(n->list_owner, -1);

    /* 清除 SCHED 状态位；若有 MISSED，则保留 SCHED 触发再次轮询 */
    val = READ_ONCE(n->state);
    do {
        new = val & ~(NAPIF_STATE_MISSED | NAPIF_STATE_SCHED | ...);
        new |= (val & NAPIF_STATE_MISSED) / NAPIF_STATE_MISSED *
                                            NAPIF_STATE_SCHED;
    } while (!try_cmpxchg(&n->state, &val, new));

    if (ret) {
        /* 重新开启硬件中断（由驱动在 poll 返回后执行） */
    }
    return ret;
}
EXPORT_SYMBOL(napi_complete_done);

napi_defer_hard_irqs 是一个优化：当 dev->napi_defer_hard_irqs > 0 时，napi_complete_done 不会立即重开中断，而是用 hrtimer 在 gro_flush_timeout 纳秒后再刷出 GRO 并重开中断，适合高速场景下减少中断频率。

2.7 netif_napi_add_weight：驱动注册 NAPI 实例

驱动在 ndo_open 中为每个 RX 队列调用 netif_napi_add() 注册 NAPI 实例（netif_napi_add 是内联包装，默认 weight 为 NAPI_POLL_WEIGHT = 64）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// net/core/dev.c: 6366
void netif_napi_add_weight(struct net_device *dev, struct napi_struct *napi,
                           int (*poll)(struct napi_struct *, int), int weight)
{
    INIT_LIST_HEAD(&napi->poll_list);
    hrtimer_init(&napi->timer, CLOCK_MONOTONIC, HRTIMER_MODE_REL_PINNED);
    napi->timer.function = napi_watchdog;   /* GRO flush 超时回调 */
    init_gro_hash(napi);
    napi->skb = NULL;
    INIT_LIST_HEAD(&napi->rx_list);
    napi->rx_count = 0;
    napi->poll = poll;          /* 驱动提供的轮询函数 */
    napi->weight = weight;      /* per-poll 最大包数（budget） */
    napi->dev = dev;
    napi->list_owner = -1;
    set_bit(NAPI_STATE_SCHED, &napi->state);
    set_bit(NAPI_STATE_NPSVC, &napi->state);
    list_add_rcu(&napi->dev_list, &dev->napi_list);  /* 加入设备的 napi_list */
    napi_hash_add(napi);        /* 加入全局 napi_hash，用于 busy-poll 查找 */
    /* 若设备开启了 threaded NAPI，为此 NAPI 创建内核线程 */
    if (dev->threaded && napi_kthread_create(napi))
        dev->threaded = 0;
}
EXPORT_SYMBOL(netif_napi_add_weight);

napi_disable() 是其对应的注销操作，内部通过自旋等待确保当前 poll 完全退出，再清除 NAPI_STATE_SCHED 位，因此可能睡眠（might_sleep() 标注），只能在进程上下文（如 ndo_stop）中调用。

2.8 softnet_data：per-CPU 的网络软状态

每个 CPU 维护一个 struct softnet_data（通过 this_cpu_ptr(&softnet_data) 访问），这是 NAPI 调度机制的核心数据结构：

1
2
3
4
5
6
7
8
9
10
11
12
// 关键字段（include/linux/netdevice.h）
struct softnet_data {
    struct list_head    poll_list;      /* 待轮询的 napi_struct 链表 */
    struct sk_buff_head process_queue;  /* 正在处理的 backlog 队列 */
    unsigned int        processed;      /* 已处理包数（统计用） */
    unsigned int        time_squeeze;   /* budget 耗尽次数 */
    ...
    bool                in_net_rx_action;  /* 是否在 net_rx_action 中 */
    bool                in_napi_threaded_poll;
    struct sk_buff_head input_pkt_queue; /* 收到的待处理 skb 队列 */
    struct napi_struct  backlog;         /* 非 NAPI 驱动的 backlog NAPI 实例 */
};

time_squeeze 是非常重要的性能指标：它记录了 net_rx_action 因 budget 耗尽或超时而提前退出的次数。该计数可在 /proc/net/softnet_stat 的第三列（十六进制）中读取，若持续增长，说明 CPU 跟不上包到达速率，需要增大 netdev_budget 或增加处理 CPU。

3.1 从 napi_gro_receive 到协议栈

驱动 poll 函数处理每一帧后，通常调用 napi_gro_receive()（见第五节），最终殊途同归地通过 netif_receive_skb() 将 sk_buff 交给网络层。调用链如下：

1
2
3
4
5
6
7
8
驱动 poll()
  └─ napi_gro_receive()
       └─ napi_skb_finish()
            └─ gro_normal_one() → gro_normal_list()
                 └─ netif_receive_skb_list_internal()
                      └─ __netif_receive_skb()
                           └─ __netif_receive_skb_one_core()
                                └─ __netif_receive_skb_core()

3.2 __netif_receive_skb_core：协议分发核心

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
// net/core/dev.c: 5279
static int __netif_receive_skb_core(struct sk_buff **pskb, bool pfmemalloc,
                                    struct packet_type **ppt_prev)
{
    struct packet_type *ptype, *pt_prev;
    rx_handler_func_t *rx_handler;
    struct sk_buff *skb = *pskb;
    ...

    net_timestamp_check(!READ_ONCE(netdev_tstamp_prequeue), skb);
    trace_netif_receive_skb(skb);

    skb_reset_network_header(skb);
    ...

another_round:
    skb->skb_iif = skb->dev->ifindex;
    __this_cpu_inc(softnet_data.processed);

    /* 1. Generic XDP（软件 XDP 程序） */
    if (static_branch_unlikely(&generic_xdp_needed_key)) {
        ret2 = do_xdp_generic(rcu_dereference(skb->dev->xdp_prog), skb);
        if (ret2 != XDP_PASS) { ret = NET_RX_DROP; goto out; }
    }

    /* 2. VLAN 去标签 */
    if (eth_type_vlan(skb->protocol)) {
        skb = skb_vlan_untag(skb);
        ...
    }

    /* 3. ptype_all：全局抓包监听（如 tcpdump AF_PACKET socket） */
    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;
        }
    }

    /* 4. 设备私有 ptype_all（设备级抓包） */
    list_for_each_entry_rcu(ptype, &skb->dev->ptype_all, list) { ... }

    /* 5. rx_handler（VLAN、bonding、bridge 等虚拟设备钩子） */
    rx_handler = rcu_dereference(skb->dev->rx_handler);
    if (rx_handler) {
        switch (rx_handler(&skb)) {
        case RX_HANDLER_ANOTHER:  goto another_round;
        case RX_HANDLER_CONSUMED: ret = NET_RX_SUCCESS; goto out;
        ...
        }
    }

    /* 6. ptype_base：按协议类型分发（ETH_P_IP → ip_rcv，ETH_P_IPV6 → ipv6_rcv...） */
    type = skb->protocol;
    deliver_ptype_list_skb(skb, &pt_prev, orig_dev, type,
                           &ptype_base[ntohs(type) & PTYPE_HASH_MASK]);
    ...
}

ptype_base 是一个大小为 PTYPE_HASH_SIZE（16）的哈希表（net/core/dev.c 第 159 行），以 ETH_P_* 协议号为键。IPv4 注册 ETH_P_IP，IPv6 注册 ETH_P_IPV6，ARP 注册 ETH_P_ARP 等。deliver_skb() 最终调用 pt_prev->func(skb, ...) 将 skb 传入对应的协议处理函数（如 ip_rcv()）：

1
2
3
4
5
6
7
8
9
10
// net/core/dev.c: 2166
static inline int deliver_skb(struct sk_buff *skb,
                               struct packet_type *pt_prev,
                               struct net_device *orig_dev)
{
    if (unlikely(skb_orphan_frags_rx(skb, GFP_ATOMIC)))
        return -ENOMEM;
    refcount_inc(&skb->users);
    return pt_prev->func(skb, skb->dev, pt_prev, orig_dev);
}

3.3 RSS 与多队列接收

现代高速网卡通过 RSS（Receive Side Scaling）在多个 RX 队列间分散流量。网卡固件计算数据帧的哈希（通常基于 5 元组：源/目的 IP、端口、协议），再与 Indirection Table 映射，将同一 flow 的包始终路由到同一队列，由对应 CPU 处理，避免了跨核的数据竞争，也保证了 TCP 流内的有序处理。

在 Linux 内核侧，每个 RX 队列对应一个 napi_struct 和一个 IRQ 向量，通过 /proc/irq/<irq_number>/smp_affinity 绑定到指定 CPU。驱动初始化时通常调用 netif_napi_add() 为每个队列注册一个 NAPI 实例，并将其挂入 dev->napi_list。

RPS（Receive Packet Steering）是 RSS 的软件实现，对不支持多队列的网卡同样适用，通过内核在 __netif_receive_skb_core() 之前的 get_rps_cpu() 将 skb 分配到合适的 CPU 处理。

3.4 sk_buff 的生命周期（RX 视角）

理解 RX 路径还需要了解 sk_buff 的生命周期管理。驱动在 ndo_open 时为每个 RX descriptor 预分配好 sk_buff（或 page fragment），并将其物理地址写入 descriptor ring 交给 DMA 引擎。当一帧到达时，硬件将数据写入预分配的内存，并标记 descriptor 为”完成”。

驱动 poll 函数检测到完成的 descriptor 后：

  1. 从 descriptor 中取出 DMA 地址，调用 dma_unmap_single() 解除 DMA 映射
  2. 将 sk_buff 的 data 指针和 len 字段设置好
  3. 调用 eth_type_trans(skb, dev) 解析以太网头，设置 skb->protocol
  4. 将硬件填写的校验和信息存入 skb->ip_summed（若有 RXCSUM offload）
  5. 调用 napi_gro_receive() 或 netif_receive_skb() 将 skb 交给协议栈
  6. 为该 descriptor 重新分配一个新的 sk_buff，维持 ring 的满足状态

sk_buff 在协议栈中向上传递时，各层通过移动 data 指针（skb_pull()）来”消费”头部，数据本身不做拷贝，直到最终被用户态 recv() 读走或被 kfree_skb() 释放。

四、发送路径（TX Path）

4.1 dev_queue_xmit → __dev_queue_xmit

上层协议（如 TCP）通过 dev_queue_xmit() 将 sk_buff 投入发送，实际工作由 __dev_queue_xmit() 完成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
// net/core/dev.c: 4150
int __dev_queue_xmit(struct sk_buff *skb, struct net_device *sb_dev)
{
    struct net_device *dev = skb->dev;
    struct netdev_queue *txq = NULL;
    struct Qdisc *q;
    int rc = -ENOMEM;

    skb_reset_mac_header(skb);
    skb_assert_len(skb);

    rcu_read_lock_bh();

    skb_update_prio(skb);
    qdisc_pkt_len_init(skb);

#ifdef CONFIG_NET_EGRESS
    /* tc egress hook（eBPF / netfilter egress） */
    if (static_branch_unlikely(&egress_needed_key)) {
        skb = sch_handle_egress(skb, &rc, dev);
        if (!skb) goto out;
    }
#endif

    /* 选择发送队列 */
    if (!txq)
        txq = netdev_core_pick_tx(dev, skb, sb_dev);

    q = rcu_dereference_bh(txq->qdisc);

    trace_net_dev_queue(skb);
    if (q->enqueue) {
        rc = __dev_xmit_skb(skb, q, dev, txq);  /* 走 qdisc 路径 */
        goto out;
    }

    /* 无队列设备（loopback、tunnel 等）的快速路径 */
    if (dev->flags & IFF_UP) {
        ...
        skb = dev_hard_start_xmit(skb, dev, txq, &rc);
        ...
    }
out:
    rcu_read_unlock_bh();
    return rc;
}
EXPORT_SYMBOL(__dev_queue_xmit);

4.2 netdev_core_pick_tx：发送队列选择

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// net/core/dev.c: 4101
struct netdev_queue *netdev_core_pick_tx(struct net_device *dev,
                                         struct sk_buff *skb,
                                         struct net_device *sb_dev)
{
    int queue_index = 0;

    if (dev->real_num_tx_queues != 1) {
        const struct net_device_ops *ops = dev->netdev_ops;

        if (ops->ndo_select_queue)
            queue_index = ops->ndo_select_queue(dev, skb, sb_dev);
        else
            queue_index = netdev_pick_tx(dev, skb, sb_dev);

        queue_index = netdev_cap_txqueue(dev, queue_index);
    }
    ...
    return netdev_get_tx_queue(dev, queue_index);
}

netdev_pick_tx() 的优先级策略：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// net/core/dev.c: 4074
u16 netdev_pick_tx(struct net_device *dev, struct sk_buff *skb,
                   struct net_device *sb_dev)
{
    struct sock *sk = skb->sk;
    int queue_index = sk_tx_queue_get(sk);  /* 1. socket 缓存的队列 */

    if (queue_index < 0 || skb->ooo_okay ||
        queue_index >= dev->real_num_tx_queues) {
        int new_index = get_xps_queue(dev, sb_dev, skb); /* 2. XPS（CPU→queue 亲和性） */

        if (new_index < 0)
            new_index = skb_tx_hash(dev, sb_dev, skb);   /* 3. 流 hash */

        if (queue_index != new_index && sk && sk_fullsock(sk) &&
            rcu_access_pointer(sk->sk_dst_cache))
            sk_tx_queue_set(sk, new_index);               /* 更新 socket 缓存 */

        queue_index = new_index;
    }
    return queue_index;
}

XPS（Transmit Packet Steering）允许将特定 CPU 的发送流量固定映射到特定 TX 队列，减少多核争用同一队列锁的开销。

4.3 qdisc 与 sch_direct_xmit

__dev_xmit_skb() 根据 qdisc 类型走不同路径：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
// net/core/dev.c: 3779
static inline int __dev_xmit_skb(struct sk_buff *skb, struct Qdisc *q,
                                  struct net_device *dev,
                                  struct netdev_queue *txq)
{
    ...
    if (q->flags & TCQ_F_NOLOCK) {
        /* nolock qdisc（如 fq_codel 的 lockless 模式） */
        if (q->flags & TCQ_F_CAN_BYPASS && nolock_qdisc_is_empty(q) &&
            qdisc_run_begin(q)) {
            /* 队列为空且抢到运行权：直接发送，绕过入队 */
            if (sch_direct_xmit(skb, q, dev, txq, NULL, true) &&
                !nolock_qdisc_is_empty(q))
                __qdisc_run(q);
            qdisc_run_end(q);
            return NET_XMIT_SUCCESS;
        }
        rc = dev_qdisc_enqueue(skb, q, &to_free, txq);
        qdisc_run(q);
        ...
    }
    /* 有锁 qdisc（默认 pfifo_fast） */
    ...
    if (contended)
        spin_lock(&q->busylock);
    spin_lock(root_lock);
    if (unlikely(test_bit(TC_QUEUE_STOPPED_BIT, &txq->state))) {
        /* TX 队列已停止，直接入队，等待唤醒 */
        rc = dev_qdisc_enqueue(skb, q, &to_free, txq);
    } else if (!qdisc_is_running(q)) {
        /* 队列空闲：直接调度发送 */
        if (sch_direct_xmit(skb, q, dev, txq, root_lock, true)) {
            __qdisc_run(q);
        }
    } else {
        /* qdisc 正在运行，入队 */
        rc = dev_qdisc_enqueue(skb, q, &to_free, txq);
    }
    ...
}

4.4 dev_hard_start_xmit：最终发送

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// net/core/dev.c: 3584
struct sk_buff *dev_hard_start_xmit(struct sk_buff *first, struct net_device *dev,
                                    struct netdev_queue *txq, int *ret)
{
    struct sk_buff *skb = first;
    int rc = NETDEV_TX_OK;

    while (skb) {
        struct sk_buff *next = skb->next;

        skb_mark_not_on_list(skb);
        rc = xmit_one(skb, dev, txq, next != NULL);
        if (unlikely(!dev_xmit_complete(rc))) {
            skb->next = next;
            goto out;
        }

        skb = next;
        if (netif_tx_queue_stopped(txq) && skb) {
            rc = NETDEV_TX_BUSY;
            break;
        }
    }
out:
    *ret = rc;
    return skb;
}

xmit_one() 内部调用 netdev_start_xmit() → __netdev_start_xmit() → ops->ndo_start_xmit(skb, dev)，至此进入驱动代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// net/core/dev.c: 3567
static int xmit_one(struct sk_buff *skb, struct net_device *dev,
                    struct netdev_queue *txq, bool more)
{
    unsigned int len;
    int rc;

    if (dev_nit_active(dev))
        dev_queue_xmit_nit(skb, dev);  /* 通知抓包 tap（tcpdump） */

    len = skb->len;
    trace_net_dev_start_xmit(skb, dev);
    rc = netdev_start_xmit(skb, dev, txq, more);
    trace_net_dev_xmit(skb, rc, dev, len);

    return rc;
}

注意 dev_nit_active() 的调用：它检查 ptype_all 或 dev->ptype_all 是否有监听者（即是否有 tcpdump/AF_PACKET 在抓包）。若有，则通过 dev_queue_xmit_nit() 将 skb 的克隆发给这些监听者，这就是为什么 tcpdump 能捕获即将发出的包，且看到的是发送前的数据。

驱动将 sk_buff 中的数据指针写入 TX descriptor ring，设置 DMA 地址，更新 tail pointer（向硬件提交新 descriptor），硬件即开始 DMA 传输。ndo_start_xmit 返回值语义如下：

  • NETDEV_TX_OK：成功，sk_buff 的所有权已转交驱动（驱动负责释放）
  • NETDEV_TX_BUSY：队列暂满，内核会停止该队列（netif_tx_stop_queue()），等驱动调用 netif_tx_wake_queue() 唤醒

发送完成后，硬件触发 TX 完成中断（或在下一次 NAPI poll 中检测），驱动在 TX 清理函数中调用 dev_consume_skb_any() 释放 sk_buff，并将 descriptor 标记为空闲以唤醒队列。

4.5 TX 路径中的 TSO 分片

当应用发送超过 MTU 的 TCP 数据且网卡支持 NETIF_F_TSO 时，内核不会在软件层面将其拆分成多个小包，而是将整个大 skb（可达 64KB，由 gso_size 约束）直接交给驱动。skb 的 skb_shinfo->gso_size 记录期望的分段大小（通常等于 TCP MSS），网卡固件负责在 DMA 时完成实际分片并填写每个分片的 TCP/IP 头，CPU 完全不参与分片计算，极大地降低了发送大文件时的 CPU 占用。

若网卡不支持 TSO，validate_xmit_skb() 会在 dev_hard_start_xmit 之前调用 skb_gso_segment() 在软件层完成分段，代价是更高的 CPU 消耗。这就是为什么关闭 TSO（ethtool -K eth0 tso off）会导致高带宽场景下 CPU 利用率显著上升。

五、GRO（Generic Receive Offload）

5.1 GRO 的设计目标

TSO 卸载了发送端的分段工作，GRO 则是其对称操作：在接收端将多个属于同一 TCP/UDP 流的小包合并成一个大包，再递交给上层协议。这大幅减少了 ip_rcv()、tcp_rcv() 等函数被调用的次数，降低了协议栈处理的 CPU 开销，在高吞吐场景下非常关键。

GRO 工作在 NAPI poll 上下文中，每个 napi_struct 维护 gro_hash 哈希表存储待合并的 skb 链。

5.2 napi_gro_receive 与 dev_gro_receive

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// net/core/gro.c: 648
gro_result_t napi_gro_receive(struct napi_struct *napi, struct sk_buff *skb)
{
    gro_result_t ret;

    skb_mark_napi_id(skb, napi);
    trace_napi_gro_receive_entry(skb);

    skb_gro_reset_offset(skb, 0);

    ret = napi_skb_finish(napi, skb, dev_gro_receive(napi, skb));
    trace_napi_gro_receive_exit(ret);

    return ret;
}
EXPORT_SYMBOL(napi_gro_receive);

核心合包逻辑在 dev_gro_receive() 中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
// net/core/gro.c: 482
static enum gro_result dev_gro_receive(struct napi_struct *napi, struct sk_buff *skb)
{
    u32 bucket = skb_get_hash_raw(skb) & (GRO_HASH_BUCKETS - 1);
    struct gro_list *gro_list = &napi->gro_hash[bucket];
    struct list_head *head = &offload_base;
    struct packet_offload *ptype;
    __be16 type = skb->protocol;
    struct sk_buff *pp = NULL;
    enum gro_result ret;
    int same_flow;

    if (netif_elide_gro(skb->dev))
        goto normal;   /* 设备禁用 GRO，直接正常处理 */

    gro_list_prepare(&gro_list->list, skb);  /* 预处理：标记同流的候选 skb */

    rcu_read_lock();
    /* 在 offload_base 中找对应协议的 GRO offload handler */
    list_for_each_entry_rcu(ptype, head, list) {
        if (ptype->type == type && ptype->callbacks.gro_receive)
            goto found_ptype;
    }
    rcu_read_unlock();
    goto normal;

found_ptype:
    skb_set_network_header(skb, skb_gro_offset(skb));
    ...
    NAPI_GRO_CB(skb)->count = 1;

    /* 调用协议 GRO handler（如 inet_gro_receive / ipv6_gro_receive） */
    pp = INDIRECT_CALL_INET(ptype->callbacks.gro_receive,
                            ipv6_gro_receive, inet_gro_receive,
                            &gro_list->list, skb);

    rcu_read_unlock();

    same_flow = NAPI_GRO_CB(skb)->same_flow;
    ret = NAPI_GRO_CB(skb)->free ? GRO_MERGED_FREE : GRO_MERGED;

    if (pp) {  /* 有已经可以 flush 的聚合包 */
        skb_list_del_init(pp);
        napi_gro_complete(napi, pp);  /* 将聚合包推入正常接收路径 */
        gro_list->count--;
    }

    if (same_flow)
        goto ok;   /* 合入已有流，当前 skb 被消费 */

    if (NAPI_GRO_CB(skb)->flush)
        goto normal;   /* 不可合并，走正常路径 */

    /* 新流：加入 gro_hash 列表 */
    ...
    gro_list->count++;
    ...
    ret = GRO_HELD;
    goto ok;

normal:
    ret = GRO_NORMAL;
ok:
    ...
    return ret;
}

napi_skb_finish() 根据 dev_gro_receive() 的返回值决定后续行为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// net/core/gro.c: 621
static gro_result_t napi_skb_finish(struct napi_struct *napi,
                                    struct sk_buff *skb,
                                    gro_result_t ret)
{
    switch (ret) {
    case GRO_NORMAL:
        gro_normal_one(napi, skb, 1);  /* 放入 napi->rx_list，批量提交 */
        break;
    case GRO_MERGED_FREE:
        /* skb 已被合并，header 被偷走，释放 skb 本身 */
        __napi_kfree_skb(skb, SKB_CONSUMED);
        break;
    case GRO_HELD:
    case GRO_MERGED:
    case GRO_CONSUMED:
        break;
    }
    return ret;
}

GRO 聚合的包在 napi_complete_done() 中通过 napi_gro_flush() 统一提交，或在聚合包超出阈值（MAX_GRO_SKBS = 8，net/core/gro.c 第 7 行）时提前 flush。聚合后的超大 skb 携带 skb_shinfo->gso_* 元数据，TCP 层可直接将其视为一个大段处理。

5.3 GRO 与 LRO 的区别

GRO 是 LRO（Large Receive Offload）的软件改进版。LRO 的实现直接在驱动层合并 TCP payload，其问题在于：合并后无法再参与 tc/netfilter 等内核网络处理，且对转发场景（路由器）有害——合并后的超大包无法被直接转发。

GRO 在软件层工作，合并时保留了各层的 header 信息和控制字段，因此合并后的包仍可正常经过 netfilter、tc 等处理，并通过 skb_shinfo->gso_* 标记通知下游（如 TCP 层）这实际是多个小包的聚合体，在必要时（如再次进入 TX 路径）可重新分段。这种设计使 GRO 在正确性和兼容性上远优于 LRO。

5.4 GRO 的边界条件

并非所有包都能被 GRO 合并，以下情况会导致 GRO 跳过（走 GRO_NORMAL 路径直接入栈）：

  • 设备调用了 netif_set_gro_max_size(dev, 0) 禁用 GRO
  • NETIF_F_GRO 特性未启用（netif_elide_gro() 返回 true）
  • skb 携带 frag_list（skb_has_frag_list(skb) 为 true）
  • skb 是 GSO 包但类型不支持（非纯 TCP GSO）
  • 协议层的 gro_receive 回调决定不合并（如 TCP 序号不连续、窗口更新包等）

对于 UDP，从 Linux 5.0 起通过 NETIF_F_GRO_UDP_FWD 引入了 UDP GRO，专为 UDP 隧道（VXLAN、Geneve）转发场景优化。

六、诊断方法

6.1 ethtool：网卡统计与队列深度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 查看网卡详细统计（对应驱动的 ndo_get_stats64 / ethtool get_ethtool_stats）
ethtool -S eth0

# 查看 RX/TX ring buffer 深度
ethtool -g eth0
# 输出示例：
# Ring parameters for eth0:
# Pre-set maximums:
# RX:  4096    TX:  4096
# Current hardware settings:
# RX:  512     TX:  512

# 调整 ring 大小（增大可减少丢包，但增加延迟）
ethtool -G eth0 rx 4096 tx 4096

# 查看 GRO/TSO/RXCSUM 等特性是否开启
ethtool -k eth0

# 开启/关闭 GRO
ethtool -K eth0 gro on

ring buffer 过小是生产环境丢包的常见原因。当 ethtool -S 输出的 rx_missed_errors 或 rx_fifo_errors 持续增长，通常意味着 ring 已满，需要增大深度或优化 NAPI 处理速度。

6.2 /proc/net/dev 与 sysfs 统计

1
2
3
4
5
6
7
8
9
10
11
12
13
# 接口级收发统计（来自内核 net_device->stats 或 ndo_get_stats64）
cat /proc/net/dev

# sysfs 接口（单个计数器，适合脚本采集）
cat /sys/class/net/eth0/statistics/rx_bytes
cat /sys/class/net/eth0/statistics/rx_dropped
cat /sys/class/net/eth0/statistics/tx_errors

# softirq 处理统计：每列分别是 total/dropped/time_squeeze/...
cat /proc/net/softnet_stat
# time_squeeze（第三列）非零说明 softirq budget 不足，考虑增大 netdev_budget
sysctl net.core.netdev_budget=600
sysctl net.core.netdev_budget_usecs=4000

6.3 中断分布与 CPU 亲和性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 查看 IRQ 分布，找到网卡相关的 IRQ 号
cat /proc/interrupts | grep eth0

# 查看某 IRQ 的 CPU 亲和性（十六进制位图）
cat /proc/irq/32/smp_affinity

# 手动绑定 IRQ 32 到 CPU 2（位图 0x4）
echo 4 > /proc/irq/32/smp_affinity

# 使用 irqbalance 自动均衡（但高性能场景建议手动绑定后禁用 irqbalance）
systemctl stop irqbalance

# 脚本：自动将网卡 irq 绑定到对应编号的 CPU（RSS 最优实践）
for i in $(seq 0 $((num_queues - 1))); do
    irq=$(cat /sys/class/net/eth0/device/msi_irqs/$i)
    echo $((1 << i)) > /proc/irq/$irq/smp_affinity
done

合理的 IRQ 亲和性配置应使 RX 队列 n 的中断、NAPI poll、以及处理该队列数据的上层线程，三者运行在同一个 CPU 或同一 NUMA 节点的 CPU 上，以最大化 cache 局部性。

6.4 用 bpftrace 追踪 NAPI 延迟

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 追踪 napi_poll 的 work 分布（每次 poll 处理了多少包）
bpftrace -e '
tracepoint:net:napi_poll {
    @work = hist(args->work);
    @budget = hist(args->budget);
}
interval:s:5 { print(@work); print(@budget); clear(@work); clear(@budget); }
'

# 测量从硬件中断触发到 napi_poll 开始的延迟（IRQ → softirq 延迟）
bpftrace -e '
kprobe:__napi_schedule { @ts[tid] = nsecs; }
tracepoint:net:napi_poll /tid != 0 && @ts[tid]/ {
    @lat_us = hist((nsecs - @ts[tid]) / 1000);
    delete(@ts[tid]);
}
interval:s:5 { print(@lat_us); clear(@lat_us); }
'

# 监控 net_rx_action 的 time_squeeze（budget 耗尽频率）
bpftrace -e '
kprobe:net_rx_action { @start[tid] = nsecs; }
kretprobe:net_rx_action /@start[tid]/ {
    @duration_us = hist((nsecs - @start[tid]) / 1000);
    delete(@start[tid]);
}
interval:s:5 { print(@duration_us); clear(@duration_us); }
'

当 @work 直方图中大量样本落在等于 budget 的位置时，说明每次 poll 都消耗了全部 budget 但队列未清空，是需要增大 netdev_budget 的信号。当 @lat_us 的 P99 延迟过高时，则应检查 CPU 是否存在 softirq 争用或 NUMA 跨节点访问问题。

6.5 典型性能问题排查清单

以下是一份基于本文所述路径的常见网络性能问题快速排查清单：

症状 可能原因 排查命令
rx_dropped 持续增长 RX ring buffer 太小，NAPI 跟不上 ethtool -g eth0，适当增大 RX ring
time_squeeze 增长 netdev_budget 不足，softirq CPU 饱和 cat /proc/net/softnet_stat，增大 netdev_budget
单核 CPU 100%（softirq） IRQ 未均衡，所有包走单 CPU cat /proc/interrupts，手动绑核
高延迟但吞吐不差 GRO 超时 flush 过大 调小 gro_flush_timeout
大流量下 TX 丢包 TX ring 太小，qdisc 队列满 ethtool -g eth0，tc -s qdisc show
tx_errors 增长 硬件 TX 超时，驱动 ndo_tx_timeout 触发 `ethtool -S eth0
NUMA 跨节点访问高 IRQ 亲和性配置不当 numastat -n，重新绑定 IRQ 到本地 NUMA node

七、总结：收发路径全貌

至此，我们可以完整描绘网卡收发包的两条核心路径：

接收路径（RX）：

1
2
3
4
5
6
7
NIC DMA → RX ring buffer
  → 硬件中断 ISR → napi_schedule() → NET_RX_SOFTIRQ
  → net_rx_action() → napi_poll() → 驱动 poll()
  → napi_gro_receive() → dev_gro_receive()（GRO 聚合）
  → napi_complete_done()（队列清空，重开中断）
  → netif_receive_skb() → __netif_receive_skb_core()
  → deliver_skb() → ip_rcv() / ipv6_rcv() → ...

发送路径（TX）：

1
2
3
4
5
6
7
8
tcp_sendmsg() → ip_output() → dev_queue_xmit()
  → __dev_queue_xmit()
  → netdev_core_pick_tx()（XPS/hash 选队列）
  → __dev_xmit_skb()（qdisc 入队）
  → sch_direct_xmit() / __qdisc_run()
  → dev_hard_start_xmit()
  → xmit_one() → ndo_start_xmit()（驱动写 TX ring）
  → 硬件 DMA → 网线

理解这两条路径的细节，是优化网络性能、排查丢包与高延迟问题的基础。下一篇将进入 IP 层，分析 ip_rcv()、路由查找（FIB）与 netfilter hook 的工作机制。

参考源文件：

  • include/linux/netdevice.h：net_device、net_device_ops、napi_struct
  • include/linux/netdev_features.h：NETIF_F_* 特性标志
  • net/core/dev.c：NAPI 调度、RX/TX 核心路径
  • net/core/gro.c：GRO 聚合逻辑

内核版本：Linux 6.4-rc1（commit ac9a78681b92）