深入理解Linux网络随笔（一）：内核是如何接收网络包的（下篇）

3、接收网络数据

3.1.1硬中断处理

数据帧从网线到达网卡时候，首先到达网卡的接收队列，网卡会在初始化时分配给自己的RingBuffer中寻找可用内存位置，寻找成功后将数据帧DMA到网卡关联的内存里，DMA操作完成后，网卡会向CPU发起一个硬中断，通知CPU有数据到达。

启动网卡到硬中断注册处理函数调用流程ign_open-->igb_request_irq-->igb_request_msix-->igb_msix_ring。

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机制napi_schedule(&q_vector->napi);return IRQ_HANDLED;
}

napi_schedule将q_vector关联的NAPI结构添加于调度队列，函数调用关系napi_schedule-->__napi_schedule-->____napi_schedule。

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

基于软中断的NAPI处理调用list_add_tail修改Per-CPU变量的softnet_data的poll_list，将驱动napi_struct传入的poll_list添加于软中断的poll_list，触发NET_RX_SOFTIRQ类型软中断。

void __raise_softirq_irqoff(unsigned int nr)
{//禁中断lockdep_assert_irqs_disabled();//追踪软中断trace_softirq_raise(nr);//触发or_softirq_pending(1UL << nr);
}
#define or_softirq_pending(x)  (S390_lowcore.softirq_pending |= (x))

通过or操作符将软中断nr对应的位设置为1，调用or_softirq_pending将该标志位添加于软中断挂起队列，触发软中断。

3.1.2软中断处理

前文分析软中断处理通过ksfortirq内核线程处理，会调用两个函数ksoftirqd_should_run和run_ksoftirqd，均调用local_softirq_pending进行处理。

#define local_softirq_pending() (S390_lowcore.softirq_pending)
static int ksoftirqd_should_run(unsigned int cpu)
{return local_softirq_pending();
}
static void run_ksoftirqd(unsigned int cpu)
{ksoftirqd_run_begin();if (local_softirq_pending()) {/** We can safely run softirq on inline stack, as we are not deep* in the task stack here.*/__do_softirq();ksoftirqd_run_end();cond_resched();return;}ksoftirqd_run_end();
}

硬中断处理是由硬件中断服务例程（ISR）触发的，调用local_softirq_pending标记软中断挂起状态，真正的中断处理由ksfortirq内核线程处理，函数调用逻辑run_ksoftirqd-->__do_softirq。

asmlinkage __visible void __softirq_entry __do_softirq(void)
{.....// 获取当前 CPU 上待处理的软中断类型的掩码pending = local_softirq_pending();// 遍历所有待处理的软中断while ((softirq_bit = ffs(pending))) {unsigned int vec_nr;int prev_count;// 将指针 'h' 移动到当前待处理软中断类型的处理函数h += softirq_bit - 1;// 计算当前软中断处理函数在软中断处理数组中的索引vec_nr = h - softirq_vec;  // 获取当前任务的预占用计数prev_count = preempt_count();// 统计当前软中断类型的处理次数kstat_incr_softirqs_this_cpu(vec_nr);// 调用 trace 函数跟踪软中断的进入trace_softirq_entry(vec_nr);// 执行软中断的处理函数h->action(h);// 调用 trace 函数跟踪软中断的退出trace_softirq_exit(vec_nr);......// 处理下一个软中断类型h++;// 右移 pending 位图，检查下一个待处理的软中断pending >>= softirq_bit;}
}

__do_softirq根据传入的软中断类型处理所有挂起的软中断，通过h->action(h)执行具体的软中断处理函数。硬中断中的设置软中断标记，和ksoftirqd中的判断是否有软中断到达，都是基于smp_processor_id()的。只要硬中断在哪个CPU上被响应，那么软中断也是在这个CPU上处理的，针对Linux软中断消耗集中一个核现象，方法：调整硬中断CPU亲和性，硬中断打散于不同核上。

设备初始化时调用open_softirq(NET_RX_SOFTIRQ, net_rx_action)，将网络接收软中断 NET_RX_SOFTIRQ 绑定到 net_rx_action 处理函数，收到软中断类型NET_RX_SOFTIRQ会调用net_rx_action接收软中断，处理网络数据包。

static __latent_entropy void net_rx_action(struct softirq_action *h)
{struct softnet_data *sd = this_cpu_ptr(&softnet_data);  // 获取当前 CPU 的软中断数据unsigned long time_limit = jiffies +usecs_to_jiffies(READ_ONCE(netdev_budget_usecs));  // 计算软中断的超时时间int budget = READ_ONCE(netdev_budget);  // 获取软中断的处理预算（每次允许处理的最大数据包数量）LIST_HEAD(list);  // 创建链表 list，用于存储要处理的 napi 结构体LIST_HEAD(repoll);  // 创建链表 repoll，用于存储需要重新投递的 napi 结构体local_irq_disable();  // 关闭CPU硬中断list_splice_init(&sd->poll_list, &list);  // 将当前 CPU 上的 poll_list 中的元素移动到 list 中local_irq_enable();  // 重新启用本地中断for (;;) {struct napi_struct *n;skb_defer_free_flush(sd);  // 清理延迟释放的数据包if (list_empty(&list)) {  // 如果没有要处理的 napi 结构体if (!sd_has_rps_ipi_waiting(sd) && list_empty(&repoll))  // 如果没有需要处理的 RPS IPI 和 repoll 列表为空goto end;  // 结束处理break;  // 如果有需要的工作，继续处理}n = list_first_entry(&list, struct napi_struct, poll_list);  // 获取待处理的第一个 napi 结构体budget -= napi_poll(n, &repoll);  // 调用 napi_poll 处理数据包，更新剩余预算/* 如果软中断窗口已耗尽，则退出处理* 允许最多运行 2 个 jiffies，这会允许平均延迟为 1.5/HZ*/if (unlikely(budget <= 0 ||time_after_eq(jiffies, time_limit))) {sd->time_squeeze++;  // 记录时间压缩（即软中断处理超时）break;  // 退出循环}}local_irq_disable();  // 禁用本地中断// 将 repoll 和 list 的元素合并到 sd->poll_list 中list_splice_tail_init(&sd->poll_list, &list);list_splice_tail(&repoll, &list);list_splice(&list, &sd->poll_list);// 如果 poll_list 中仍有元素，重新唤起软中断if (!list_empty(&sd->poll_list))__raise_softirq_irqoff(NET_RX_SOFTIRQ);net_rps_action_and_irq_enable(sd);  // 处理 RPS（Receive Packet Steering）和启用中断
end:;
}

遍历所有待处理的网络接收软中断，核心获取当前CPU软中断数据softnet_data，list_first_entry遍历poll_list，调用 napi_poll 处理数据包，上文中分析了NAPI机制的poll函数是igb_poll。

static int igb_poll(struct napi_struct *napi, int budget)
{
.....//TX发送队列if (q_vector->tx.ring)clean_complete = igb_clean_tx_irq(q_vector, budget);//RX接收队列if (q_vector->rx.ring) {int cleaned = igb_clean_rx_irq(q_vector, budget);// 累计本次轮询处理的数据包数量work_done += cleaned;if (cleaned >= budget)clean_complete = false;}
......return work_done;
}

igb_poll是 igb 网卡驱动中 NAPI 轮询的核心函数，负责清理发送和接收队列的中断。核心处理逻辑igb_clean_rx_irq和igb_clean_tx_irq。

static int igb_clean_rx_irq(struct igb_q_vector *q_vector, const int budget)
{......rx_desc = IGB_RX_DESC(rx_ring, rx_ring->next_to_clean);  // 获取当前的接收描述符size = le16_to_cpu(rx_desc->wb.upper.length);  // 获取数据包的大小rx_buffer = igb_get_rx_buffer(rx_ring, size, &rx_buf_pgcnt);  // 获取接收缓冲区pktbuf = page_address(rx_buffer->page) + rx_buffer->page_offset;  // 获取数据包的缓冲区地址if (!skb) {unsigned char *hard_start = pktbuf - igb_rx_offset(rx_ring);  // 获取数据包的起始地址unsigned int offset = pkt_offset + igb_rx_offset(rx_ring);  // 计算数据包的偏移xdp_prepare_buff(&xdp, hard_start, offset, size, true);  // 为 XDP 准备数据包xdp_buff_clear_frags_flag(&xdp);  // 清除 XDP 的分段标志skb = igb_run_xdp(adapter, rx_ring, &xdp);  // 运行 XDP 处理，构建 skb
}if (IS_ERR(skb)) {unsigned int xdp_res = -PTR_ERR(skb);if (xdp_res & (IGB_XDP_TX | IGB_XDP_REDIR)) {  // 如果是 XDP 传输或重定向xdp_xmit |= xdp_res;igb_rx_buffer_flip(rx_ring, rx_buffer, size);  // 切换缓冲区} else {rx_buffer->pagecnt_bias++;  // 增加页面计数偏移}total_packets++;  // 增加数据包计数total_bytes += size;  // 增加字节数
}else if (skb) {igb_add_rx_frag(rx_ring, rx_buffer, skb, size);  // 将接收到的数据包添加到 skb 中
}napi_gro_receive(&q_vector->napi, skb);  // 将 skb 传递给 NAPI 进行进一步处理igb_put_rx_buffer(rx_ring, rx_buffer, rx_buf_pgcnt);  // 释放接收缓冲区if (cleaned_count)igb_alloc_rx_buffers(rx_ring, cleaned_count);......
}

igb_clean_tx_irq核心将数据帧从RingBuffer中摘下，igb_alloc_rx_buffers重新申请新的skb再重新挂起，NAPI机制下一步处理调用napi_gro_receive。

gro_result_t napi_gro_receive(struct napi_struct *napi, struct sk_buff *skb)
{......ret = napi_skb_finish(napi, skb, dev_gro_receive(napi, skb));......return ret;
}
EXPORT_SYMBOL(napi_gro_receive);

napi_gro_receive用于网卡GRO特性，合并多个小的数据包（通常是同一流的 TCP 数据包）为一个较大的数据包，从而减少协议栈的处理开销。调用napi_skb_finish完成GRO处理。

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);break;case GRO_MERGED_FREE:if (NAPI_GRO_CB(skb)->free == NAPI_GRO_FREE_STOLEN_HEAD)napi_skb_free_stolen_head(skb);else if (skb->fclone != SKB_FCLONE_UNAVAILABLE)__kfree_skb(skb);else__kfree_skb_defer(skb);break;case GRO_HELD:case GRO_MERGED:case GRO_CONSUMED:break;}return ret;
}

正常数据包处理GRO_NORMAL，需要合并的数据包处理GRO_MERGED_FREE，根据不同的方式选择不同的释放方式，函数调用逻辑gro_normal_one-->gro_normal_list-->netif_receive_skb_list_internal-->__netif_receive_skb_list-->__netif_receive_skb_core，数据包发送于协议栈。

3.1.3网络协议栈处理

static int __netif_receive_skb_core(struct sk_buff **pskb, bool pfmemalloc,struct packet_type **ppt_prev)
{......// 遍历全局的协议类型链表并传递 skb,tcpdump入口list_for_each_entry_rcu(ptype, &ptype_all, list) {if (pt_prev)ret = deliver_skb(skb, pt_prev, orig_dev);pt_prev = ptype;}type = skb->protocol;// 如果没有精确匹配，处理协议类型if (likely(!deliver_exact)) {deliver_ptype_list_skb(skb, &pt_prev, orig_dev, type,&ptype_base[ntohs(type) & PTYPE_HASH_MASK]);
}// 传递到设备特定协议链表deliver_ptype_list_skb(skb, &pt_prev, orig_dev, type,&orig_dev->ptype_specific);// 如果 skb 的设备不是原始设备，进行协议处理if (unlikely(skb->dev != orig_dev)) {deliver_ptype_list_skb(skb, &pt_prev, orig_dev, type,&skb->dev->ptype_specific);}  
}

ptype_all 是一个全局链表，包含了所有已注册的协议类型及其处理回调，在 __netif_receive_skb_core 函数中，首先会遍历这个链表，依次处理每个协议类型，并调用与协议相关的处理函数，ptype_base 是一个基于协议类型的哈希表，它包含了协议类型（如 IPv4、IPv6、TCP 等）对应的特定处理函数，当数据包的协议类型与哈希表中的某个条目匹配时，数据包会被传递到该处理函数。例如，ip_rcv 的地址通常是保存在 ptype_base 哈希表中的。

static inline int deliver_skb(struct sk_buff *skb,struct packet_type *pt_prev,struct net_device *orig_dev)
{......//调用协议处理函数 (pt_prev->func) 并将 skb 传递给它return pt_prev->func(skb, skb->dev, pt_prev, orig_dev);
}
static struct packet_type ip_packet_type __read_mostly = {.type = cpu_to_be16(ETH_P_IP),.func = ip_rcv,
};

deliver_skb 将接收到的 skb传递给协议处理函数。从 packet_type 结构体中获取 func 字段，并将 skb 数据包传递给该函数进行处理。这里的 pt_prev->func 是一个协议回调函数，指向处理该协议类型数据包的函数（例如，对于 IPv4 数据包，func 指向 ip_rcv 函数）。

3.1.4IP层处理

数据包经过协议栈处理后会被传递到IP层进行处理，收包方向IP层入口函数ip_rcv。

int ip_rcv(struct sk_buff *skb, struct net_device *dev, struct packet_type *pt,struct net_device *orig_dev)
{struct net *net = dev_net(dev);skb = ip_rcv_core(skb, net);if (skb == NULL)return NET_RX_DROP;return NF_HOOK(NFPROTO_IPV4, NF_INET_PRE_ROUTING,net, NULL, skb, dev, NULL,ip_rcv_finish);
}

接收到的数据包会经过ip_rcv_core进行基本处理，例如对数据包进行有效性检查、协议解析等，处理成功会触发IPV4数据包Netfilter钩子链，在 NF_INET_PRE_ROUTING 钩子处插入数据包处理，NF_INET_PRE_ROUTING 是所有接收数据包到达的第一个 hook 触发点，在路由判断之前执行，对应的回调函数ip_rcv_finish。

static int ip_rcv_finish(struct net *net, struct sock *sk, struct sk_buff *skb)
{struct net_device *dev = skb->dev;int ret;/* if ingress device is enslaved to an L3 master device pass the* skb to its handler for processing*/skb = l3mdev_ip_rcv(skb);if (!skb)return NET_RX_SUCCESS;ret = ip_rcv_finish_core(net, sk, skb, dev, NULL);if (ret != NET_RX_DROP)ret = dst_input(skb);return ret;
}

ip_rcv_finish负责接收和处理通过 IPv4 协议栈传输的网络数据包，核心的IP数据包处理函数调用dst_input传递数据包。

static inline int dst_input(struct sk_buff *skb)
{return INDIRECT_CALL_INET(skb_dst(skb)->input,ip6_input, ip_local_deliver, skb);
}

基于路由类型skb_dst(skb)选择对应的处理函数，通过INDIRECT_CALL_INET宏选择IPV4/IPV6协议数据包处理函数，IPV4选择调用ip_local_deliver。

int ip_local_deliver(struct sk_buff *skb)
{/**	Reassemble IP fragments.*/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);
}
EXPORT_SYMBOL(ip_local_deliver);

ip_local_deliver处理本地IPV4数据包，接收一个网络数据包skb，调用ip_is_fragment检查是否需要进行IP分片重组，对不分片/已重组的数据包调用NF_HOOK，NF_INET_LOCAL_IN处理本地接收到的数据包(并不是经过路由转发的数据包)，对应的处理函数ip_local_deliver_finish。

参考资料：《深入理解Linux网络》