Linux 内核 NAPI 网络接收机制深度工程实战：从硬中断到零拷贝的完整数据通路

# Linux 内核 NAPI 网络接收机制深度工程实战：从硬中断到零拷贝的完整数据通路 Linux 内核 NAPI（New API）是当今高性能网络栈的基石机制。从 10G 到 400G 网卡，从云原生 Service Mesh 到 DPDK 旁路方案，理解 NAPI 不仅能帮助网络应用突破性能瓶颈，更是掌握 Linux 内核网络子系统设计的关键入口。 ## 一、NAPI 的历史背景与设计哲学 在 NAPI 出现之前（Linux 2.4 及更早），网络接收完全依赖硬中断：每到达一个数据包，网卡触发 IRQ，CPU 暂停当前任务执行中断处理程序。 中断驱动模式的致命问题在于高吞吐场景下的 "活锁"（Livelock）：当包到达速率超过内核处理能力时，CPU 时间被中断处理程序耗尽，协议栈无法及时消费接收队列中的包，导致丢包率 100%，而 CPU 利用率却是 100%。 NAPI 的核心思想是混合中断 + 轮询： 1. 第一个包到达时触发中断 2. 中断 handler 关闭后续接收中断，schedule NAPI poll 3. 在 softirq 上下文中批量 poll 收包 4. 队列排空后退出 poll，重新开启接收中断 ## 二、核心数据结构 ```c struct napi_struct { struct list_head poll_list; // 挂载到 softnet_data->poll_list unsigned long state; // NAPI_STATE_SCHED / NAPI_STATE_DISABLE int weight; // 默认 64（对应 net.core.netdev_budget） unsigned int gro_count; // GRO 已合并计数 int (*poll)(struct napi_struct *napi, int budget); // 轮询函数 struct net_device *dev; // 关联网卡设备 struct sk_buff *gro_list; // GRO 待合并队列 struct sk_buff *rx_skb; // 接收侧 sk_buff struct list_head napi_hash_node; // 多队列哈希桶 }; ``` ```c struct softnet_data { struct list_head poll_list; // 待轮询 NAPI 设备列表 struct softirq_action action; // 注册的 NET_RX_SOFTIRQ handler struct Qdisc *output_queue; // 输出队列（用于 TX 路径） unsigned int received_queue_tail; unsigned int processed_weight; unsigned int input_queue_head ____cacheline_aligned_in_smp; unsigned int input_queue_tail; }; ``` ## 三、NAPI 完整生命周期 ### 3.1 注册阶段（驱动加载时） ```c // drivers/net/ethernet/intel/igb/igb_main.c static int igb_probe(struct pci_dev *pdev, const struct pci_device_id *ent) { netif_napi_add(netdev, &adapter->napi, igb_poll, 64); // weight=64: 单次 poll 最多处理 64 个包 } void netif_napi_add(struct net_device *dev, struct napi_struct *napi, int (*poll)(struct napi_struct *, int), int weight) { napi->poll = poll; napi->weight = weight; napi->state = 0; INIT_LIST_HEAD(&napi->poll_list); set_bit(NAPI_STATE_SCHED, &napi->state); // 初始状态 } ``` ### 3.2 调度入口（硬中断触发） ```c // 网卡中断 handler static irqreturn_t igb_intr(int irq, void *data) { // 关键操作：关闭 RX 中断 writel(~0, adapter->hw_addr + EIMC); // 调度 NAPI（原子操作） if (likely(napi_schedule_prep(&q_vector->napi))) __napi_schedule(&q_vector->napi); return IRQ_HANDLED; } ``` ```c // kernel/core/dev.c static inline bool napi_schedule_prep(struct napi_struct *napi) { // 检查未被 disable 且未在 poll_list 上 smp_mb__before_atomic(); if (unlikely(test_bit(NAPI_STATE_DISABLE, &napi->state))) return false; if (!test_and_set_bit(NAPI_STATE_SCHED, &napi->state)) return true; // 设置成功，尚未在 list 中 return false; } void __napi_schedule(struct napi_struct *napi) { unsigned long flags; local_irq_save(flags); list_add_tail(&napi->poll_list, &get_cpu_var(softnet_data)->poll_list); __raise_softirq_irqoff(NET_RX_SOFTIRQ); // 触发软中断 local_irq_restore(flags); } ``` ### 3.3 Softirq Poll 主循环 ```c // kernel/core/dev.c static void net_rx_action(struct softirq_action *h) { struct softnet_data *sd = this_cpu_ptr(&softnet_data); unsigned long time_limit = jiffies + 2; // 2 jiffies 超时 int budget = netdev_budget; // 默认 300 local_irq_disable(); list_splice_init(&sd->poll_list, &sd->to_flush); local_irq_enable(); while (!list_empty(&sd->to_flush)) { struct napi_struct *napi; int work, weight; napi = list_first_entry(&sd->to_flush, struct napi_struct, poll_list); budget -= weight = napi->weight; work = napi->poll(napi, weight); budget += weight; if (unlikely(work == weight)) { // budget 耗尽：gro_normal_list 将其重新挂回 poll_list if (unlikely(!list_empty(&napi->poll_list)) budget = 0; } if (budget <= 0 || work < weight || time_after_eq(jiffies, time_limit)) break; list_del_init(&napi->poll_list); napi_complete_done(napi, work); // 退出轮询模式 } } ``` ### 3.4 驱动 Poll 函数实现 ```c // drivers/net/ethernet/intel/igb/igb_main.c static int igb_poll(struct napi_struct *napi, int budget) { struct igb_q_vector *q_vector = container_of(napi, struct igb_q_vector, napi); struct igb_ring *ring = q_vector->rx_ring; int work_done = 0; // 第一步：在 RX ring buffer 上批量收包 while (work_done < budget) { union advanced_rx_desc *rx_desc; struct sk_buff *skb; u16 cleaned_count = 0; rx_desc = E1000_RX_DESC_ADC(ring, ring->next_to_clean); if (!(rx_desc->wb.upper.status_error & E1000_RXD_STAT_DD)) break; // 没有新包 // 分配 skb 并通过 DMA 映射 skb = igb_fetch_rx_buffer(ring, rx_desc, work_done); // 校验校验和 if (igb_test_staterr(rx_desc, E1000_RXD_ERR)) { dev_kfree_skb_any(skb); continue; } // 写入 skb 元数据 skb_put(skb, igb_get_pkt_len(ring, rx_desc)); skb->protocol = eth_type_trans(skb, ring->netdev); // 第二步：GRO 合并尝试 if (skb_gro_receive(&ring->napi->napi_gro_list, skb) != GRO_MERGED_FREE) { work_done++; } } // 第三步：刷新 TX ring（合并 TX 清理） igb_clean_tx_irq(q_vector, budget); // 提交 GRO 合并好的包到协议栈 gro_normal_list(napi); // 通知网卡 RX ring 有可用描述符 if (cleaned_count) igb_alloc_rx_buffers(ring, cleaned_count); return work_done; } ``` ## 四、GRO 通用接收卸载 GRO（Generic Receive Offoff）在 NAPI 收包后、送入协议栈前进行大包合并。 ### 4.1 GRO 合并检查清单 ```c // net/core/dev.c enum gro_result { GRO_MERGED, // 合并成功 GRO_MERGED_FREE, // 合并且释放了 skb GRO_HELD, // 等待后续合并 GRO_NORMAL, // 普通包，不合并 GRO_DROP // 丢弃 }; ``` ### 4.2 GRO 核心合并逻辑 ```c // net/core/dev.c struct sk_buff *skb_gro_receive(struct napi_struct *napi, struct sk_buff *skb) { struct sk_buff *p; struct sk_buff **pp = NULL; unsigned int head_offset = skb_headroom(skb); // 遍历 GRO list 中已有的包 for (pp = &napi->gro_list; (p = *pp) != NULL; pp = &p->next) { if (!NAPI_GRO_CB(p)->same_flow) continue; // 检查是否与现有包匹配（same_flow 标记） if (tcp_gro_complete(skb) == 0) { // 合并到尾部 NAPI_GRO_CB(p)->count++; skb_gro_pull(skb, offsetof(struct tcphdr, check)); goto merge; } } // 无法合并：直接挂到 GRO list 尾部 skb->next = napi->gro_list; napi->gro_list = skb; NAPI_GRO_CB(skb)->count = 1; NAPI_GRO_CB(skb)->last = skb; NAPI_GRO_CB(skb)->same_flow = 1; return skb; } ``` ## 五、多队列 NAPI 与多核分发 当网卡支持多队列（RSS: Receive Side Scaling）时，每个 RX 队列有独立的 NAPI 上下文。 ```c // 驱动创建多队列 NAPI for (i = 0; i < num_queues; i++) { struct mlx5e_rq *rq = &priv->profile->rxq[i]; netif_napi_add(priv->netdev, &rq->napi, mlx5e_poll_rx_cq, 64); napi_enable(&rq->napi); // 设置中断亲和性 irq_set_affinity_hint(priv->rx_irq[i].vector, cpumask_of(i % num_online_cpus())); } ``` ### 5.1 RSS 硬件分发流程 ``` 数据包到达 → RSS Hash（基于五元组）→ Redirection Table → 指定 RX Queue → 对应 NAPI poll ``` ### 5.2 RPS/RFS 软件分发 单队列网卡通过 RPS（Receive Packet Steering）在软件层模拟多队列： ```c // net/core/dev.c static int get_rps_cpu(struct net_device *dev, struct sk_buff *skb, struct rps_dev_flow **rflowp) { struct rps_map *map = rcu_dereference(dev->rps_map); int cpu = -1; if (map) { // 用 skb_hash 选择 CPU（与 RSS 相同哈希算法） cpu = map->cpus[recip_scale(skb_hash, map->len)]; } // RFS：根据 flow 信息选择历史 CPU if (static_key_false(&rfs_needed)) { cpu = get_rps_dev_flow_cpu(dev, skb_hash); } return cpu; } ``` ## 六、Busy Polling 对于超低延迟应用（高频交易、RDMA、5G UPF），Linux 提供了 busy polling 机制：应用层通过 `setsockopt(SOCK_RX_RING_BUSY_POLL)` 在 recvmsg 系统调用中直接轮询 NAPI poll list，绕过 softirq 调度延迟。 ```c // net/core/dev.c static int napi_busy_loop(void *arg) { struct napi_struct *napi = arg; bool (*busy_poll)(void *) = napi->poll_busy_poll; unsigned long end_time = busy_poll ? 0 : jiffies + 2; local_bh_disable(); napi->poll_busy_poll_data = NULL; napi_busy_loop_on(); // 设置 socket 的繁忙状态标志 do { if (busy_poll(busy_poll_data)) { // 成功处理了包 refcount_set(&napi->refcnt, 1); local_bh_enable(); return 1; } // 超时检查 if (end_time && time_after_eq(jiffies, end_time)) { // 超时回退到普通 softirq NAPI local_bh_enable(); return 0; } } while (!need_resched()); local_bh_enable(); return 0; } ``` 关键内核参数： - `net.core.busy_poll = 50`（微秒）：recvmsg 时忙等待的微秒数 - `net.core.busy_budget = 300`：每次 busy poll 的最大预算 - `net.core.dev_weight = 64`：NAPI poll 的默认权重 ## 七、NAPI 卸载（netpoll 和 XDP） ### 7.1 netpoll 系统进入某些不可屏蔽中断（如 kdump、crash）时，内核通过 netpoll 驱动仍能收包： ```c // net/core/netpoll.c void netpoll_poll(struct net_device *dev) { struct napi_struct *napi; list_for_each_entry(napi, &dev->napi_list, dev_list) napi->poll(napi, 64); } ``` ### 7.2 XDP 与 NAPI 协同 eBPF XDP 程序运行在 NAPI poll 路径的最前端，可绕过协议栈直接返回 XDP_DROP/XDP_PASS/XDP_REDIRECT。 ```c static int do_xdp_generic(struct net_device *dev, struct sk_buff *skb) { struct bpf_prog *xdp_prog = rcu_dereference(dev->xdp_prog); struct xdp_frame *xdpf; if (!xdp_prog) return XDP_PASS; // 没有 XDP，包继续走 NAPI/GRO // 调用 eBPF XDP 程序 act = bpf_prog_run_xdp(xdp_prog, xdp_convert_frame_to_buff(xdpf)); switch (act) { case XDP_DROP: kfree_skb(skb); return XDP_DROP; case XDP_REDIRECT: xdp_do_redirect(dev, xdp_buff, xdp_prog); return XDP_REDIRECT; case XDP_PASS: return XDP_PASS; } } ``` ## 八、性能优化与排错 ### 8.1 关键性能参数 | 参数 | 路径 | 默认值 | 说明 | |------|------|---------|------| | `netdev_budget` | /proc/sys/net/core/netdev_budget | 300 | 单次 softirq 最多处理 NAPI poll 总次数 | | `dev_weight` | napi->weight | 64 | 单队列单次 poll 最大包数 | | `netdev_max_backlog` | /proc/sys/net/core/netdev_max_backlog | 1000 | 每 CPU 输入队列最大长度 | | `netdev_tstamp_prequeue` | /proc/sys/net/core/netdev_tstamp_prequeue | 1 | 是否先快速入队再处理 | ### 8.2 常见排错场景 **场景 1：RX drop 发生在 ring buffer** ```bash # ethtool -S eth0 | grep -E "drop|miss" rx_dropped: 1523401 rx_missed_errors: 8921 # 增大 RX ring buffer ethtool -G eth0 rx 4096 tx 4096 ``` **场景 2：单个 CPU 软中断过载** ```bash # mpstat -P ALL 1 %soft 分布不均时： # 启用多队列 RSS ethtool -X eth0 equal 8 # 或手动设置 RPS echo f > /sys/class/net/eth0/queues/rx-0/rps_cpus ``` **场景 3：busy poll 延迟抖动** ```bash # 查看当前 NAPI 调度延迟 cat /proc/net/softnet_stat | column -t # 第 0 列：总处理包数 # 第 1 列：drop 数 # 第 2 列：time_squeeze（NAPI_POLL 预算耗尽次数） # 第 3 列：cpu冲突数 ``` ### 8.3 perf 分析 NAPI 路径 ```bash # NAPI poll 函数时延分布 perf record -g -a -e cycles:k -- sleep 10 perf report --sort=dso,symbol | grep napi # GRO 合并效率 bpftrace -e 'kprobe:skb_gro_receive { @ = count(); }' # RX 队列丢包原因 perf record -e skb:kfree_skb -ag -- sleep 5 ``` ## 九、实战案例：10G 网卡调优 25G Mellanox ConnectX-5 网卡 → 目标：单核线性小包（64 字节）PPS。 ``` 原始状态： rps_cpus = 0（未启用 RPS） 中断绑定 CPU0 单核 SS：2.1 MPS 中断活锁：CPU 100% 调优步骤： 1. 开启 RSS 硬件多队列： ethtool -X eth0 equal 8 2. 绑定中断到不同 CPU： for i in $(seq 0 7); do echo $(printf "%x" $((1 << i))) > /proc/irq/$irq-$i/smp_affinity done 3. 调整 NAPI weight： 在驱动中设置 netif_napi_add(..., 128) # 加大小包 poll budget 4. 调整 netdev_budget（全局）： sysctl -w net.core.netdev_budget=600 5. 开启 busy poll（低延迟应用）： sysctl -w net.core.busy_poll=300 # 300us 忙等 sysctl -w net.core.busy_budget=600 调优后： 8 CPU 核总 PPS：14.8 MPP（每核 1.85 MPP） 相比 2.1 MPS 单核：604% 提升 ``` ## 十、内核新进展 ### 10.1 napi_defer_hard_irqs（Linux 5.12+） 引入中断延迟机制：即使 NAPI 完成轮询，也等待若干包之后才重新开启中断，减少中断风暴。 ```c // net/core/dev.c static int napi_complete_done(struct napi_struct *n, int work_done) { if (DEFER_NAPI) { // 在 RX ring 中保留的包达到 irq_defer_hard_irqs 阈值 // 才真正开启中断 gro_normal_list(n); n->irq_defer_hard_irqs--; return; } // 正常开启 RX 中断 clear_bit(NAPI_STATE_SCHED, &n->state); if (test_bit(NAPI_STATE_MISSED, &n->state)) __napi_schedule(n); } ``` ### 10.2 ADQ（Application Device Queues / Intel） Intel E810 系列网卡支持的 ADQ 允许为特定应用创建专用 NAPI 通道，绕过内核协议栈直接送到 socket 接收缓冲区。 ```c // net/core/dev.c static int ndo_set_channel(struct net_device *dev, struct ethtool_channels *channel) { // 创建 PCP（Priority to Queue）映射 ice_vsi_map_rings_to_vectors(vsi); ice_req_irq_msix_mbz(vsi, "adq-vector"); return 0; } ``` ### 10.3 page_pool 与 NAPI 回收 Linux 5.3+ 引入 page_pool，实现 RX 页面的快速回收（跳过 SKB 销毁的 SLAB 路径），配合 NAPI 直接操作高速缓存。 ```c // net/core/page_pool.c void page_pool_put_page(struct page_pool *pool, struct page *page, bool allow_direct) { // 直接返回到热页缓存（不经过 put_page 和 buddy system） if (allow_direct && in_serving_softirq()) { list_add(&page->lru, &pool->ring); if (++pool->ring.producer > PP_ALLOC_CACHE_REFILL) refill_page_pool(pool); return; } put_page(page); // 慢路径 } ``` ## 总结 NAPI 代表的混合中断-轮询模型是 Linux 高性能网络的核心机制。从内核 2.6 至今，NAPI 演进了 GRO 聚合、多队列 和 busy polling，未来将与 XDP 和 page_pool 进一步深度融合，支撑起更高效、更灵活的网络栈演进。理解 NAPI 不仅是网络驱动开发的基础，更是突破高并发网络服务器性能瓶颈的关键路径。