Linux收到一个网络包是怎么处理的？

摘要

编辑

1 从网卡开始

2 硬中断，有点短

2.1 Game Over

3 接力——软中断

3.1 NET_RX_SOFTIRQ 软中断的开始

3.2 数据包到了协议栈

3.3 网络层处理

3.4 传输层处理

4 应用层的处理

5 总结

摘要

一个网络包的接收始于网卡，经层层协议栈的解析，终于应用层。今天来循着一个网络包的足迹👣，深入学习一下 Linux 下接收数据的处理流程。

文中引用 Linux 内核源码基于版本 2.6.34，并做了一些删减以提高可读性。

1 从网卡开始

三更半夜，一串二进制的比特流在错综复杂的网线中极速穿行，并顺着网线爬到了你的家中。敏锐的网卡感知到了这个不速之客的到来，将它放到了辖下的某个接收队列。

如何查看网卡 RingBuf 的大小? 借助 ethtool 工具，如下表示接收缓存区支持存放 1024 个数据帧：

[root@centos ~]# ethtool -g eth0
Ring parameters for eth0:
Pre-set maximums:
RX: 1024
RX Mini: 0
RX Jumbo: 0
TX: 1024
...

ifconfig 输出中 overruns 表示的就是因 RingBuf 满而不得已丢弃的数据帧的个数。

接着，网卡在内存中提前开辟的缓冲区—— RingBuf 中循着空闲位置，找到后，由 DMA 引擎把数据直接从网卡的接收队列拷贝至 RingBuf 中。

现代网络接口卡（NICs）通常都会内置直接内存访问（DMA）引擎。DMA是一种允许硬件设备直接向主内存读写数据的技术，而无需CPU的直接介入，这样可以显著提高数据传输的效率，减少CPU的负载。

紧接着~网卡向 CPU 同学发出了一个电信号——硬中断：“起来接客！”

2 硬中断，有点短

CPU 左脚被电了一激灵，便知道是网卡送来了好东西，便开始查硬中断注册表，找到网卡提前注册在这里的回调函数。对于 Intel 的 igb 网卡，其注册的硬中断处理函数为 igb_msix_ring 。

// drivers/net/igb/igb_main.c
static irqreturn_t igb_msix_ring(int irq, void *data)
{struct igb_q_vector *q_vector = data;// 记录硬件中断频率igb_write_itr(q_vector);// 走 napi 处理数据napi_schedule(&q_vector->napi);return IRQ_HANDLED;
}

NAPI 是 linux 内核网络子系统的一个特性，通过定期轮询的方式处理聚合后的数据，可以减少高负载下的中断次数来降低 CPU 的使用率，进而提升系统的整体性能。

顺着 napi_schedule 这个调用一路前行，最终来到了 __napi_schedule ：

void __napi_schedule(struct napi_struct *n)
{unsigned long flags;local_irq_save(flags);// 将 napi 带来的的数据帧 list 放到每 cpu 的 sofnet_data 的 list 中list_add_tail(&n->poll_list, &__get_cpu_var(softnet_data).poll_list);// 触发 NET_RX_SOFTIRQ 软中断__raise_softirq_irqoff(NET_RX_SOFTIRQ);local_irq_restore(flags);
}

这里触发软中断的方式类似于发送信号，只是简单的修改了一个变量，将软中断信号设置到了 irq_stat 中。irq_stat 是一个数组，以 cpu 号为索引。所以这里操作之后，在硬中断对应的那个 cpu 上设置了软中断标记。后续的软中断也是在同一个 cpu 上继续执行的。所以如果发现软中断负载不均的话，就需要调整硬中断的 cpu 亲和性，将其均衡到不同的 cpu 核上去。或是更换支持多队列的网卡，每个队列都会有一个自己的中断号，可以更好的支持负载均衡。

#define __raise_softirq_irqoff(nr) \do { or_softirq_pending(1UL << (nr)); } while (0)#define or_softirq_pending(x)  (local_softirq_pending() |= (x))#ifndef __ARCH_IRQ_STAT
extern irq_cpustat_t irq_stat[];		/* defined in asm/hardirq.h */
#define __IRQ_STAT(cpu, member)	(irq_stat[cpu].member)
#endif/* arch independent irq_stat fields */
#define local_softirq_pending() \__IRQ_STAT(smp_processor_id(), __softirq_pending)

2.1 Game Over

到这里，硬中断的使命就完成了。可见，其只是做了及其简单的处理：

记录硬中断频率
将待处理数据帧 list 挂到 softnet_data 的 poll_list 上
触发软中断

3 接力——软中断

再来看一下这张图，cpu 现在已经执行完了硬中断上网卡注册的回调函数，并触发了一个 NET_RX_SOFTIRQ 软中断。

在 linux 启动的时候，就已经给每个 cpu 启动了一个名为的 ksoftirq/x 的内核线程，ksoftirq/x 启动后，它的入口函数是这个：

// kernel/softirq.c
static int run_ksoftirqd(void * __bind_cpu)
{set_current_state(TASK_INTERRUPTIBLE);while (!kthread_should_stop()) {if (!local_softirq_pending()) {schedule();}__set_current_state(TASK_RUNNING);while (local_softirq_pending()) {do_softirq();}set_current_state(TASK_INTERRUPTIBLE);}__set_current_state(TASK_RUNNING);return 0;
}

local_softirq_pending 是不是挺眼熟的，前面触发软中断即是调用它来获取软中断保存的变量。在 ksoftirq/x 中会循环调用 local_softirq_pending 判断是否有待处理的软中断，没有就会schdule 出去，否则就要调用 do_softirq 开始处理软中断了。do_softirq 进一步调用了 __do_softirq：

// kernel/softirq.c
asmlinkage void __do_softirq(void)
{do {if (pending & 1) {trace_softirq_entry(h, softirq_vec);// 调用对应软中断的回调方法h->action(h);trace_softirq_exit(h, softirq_vec);}h++;pending >>= 1;} while (pending);
}

在 __do_softirq 中，会遍历软中断注册表，查找发生了软中断的回调函数 action ，调用之。对于 NET_RX_SOFTIRQ 软中断，对应的回调函数为 net_rx_action。

3.1 NET_RX_SOFTIRQ 软中断的开始

net_rx_action 中从 softnet_data 的 poll_list 中拿到待处理的数据帧，遍历这个 poll_list ，然后依次调用一个 poll 方法处理这些数据帧。

static void net_rx_action(struct softirq_action *h)
{// 从 softnet_data 的 poll_list 中获取待处理数据帧，这是前面硬中断cb中放在这里的struct list_head *list = &__get_cpu_var(softnet_data).poll_list;while (!list_empty(list)) {/* Even though interrupts have been re-enabled, this* access is safe because interrupts can only add new* entries to the tail of this list, and only ->poll()* calls can remove this head entry from the list.*/n = list_first_entry(list, struct napi_struct, poll_list);if (test_bit(NAPI_STATE_SCHED, &n->state)) {work = n->poll(n, weight);trace_napi_poll(n);}...}
}

poll 也是一个网卡驱动注册的回调方法，对于 Inter 的 igb 网卡，它是 igb_poll：

// drivers/net/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);if (q_vector->tx_ring)tx_clean_complete = igb_clean_tx_irq(q_vector);if (q_vector->rx_ring)igb_clean_rx_irq_adv(q_vector, &work_done, budget);...
}

igb_poll 方法中，做的事主要有两点：

清理发送缓冲中的无用数据
处理&清理接收缓冲中的数据

我们跟进看一下 igb_clean_rx_irq_adv 对接收的处理流程：

// drivers/net/igb/igb_main.c
static bool igb_clean_rx_irq_adv(struct igb_q_vector *q_vector,int *work_done, int budget)
{while (staterr & E1000_RXD_STAT_DD) {// 将数据包从 RingBuf 上取下来skb = buffer_info->skb;prefetch(skb->data - NET_IP_ALIGN);buffer_info->skb = NULL;...skb_record_rx_queue(skb, rx_ring->queue_index);vlan_tag = ((staterr & E1000_RXD_STAT_VP) ?le16_to_cpu(rx_desc->wb.upper.vlan) : 0);// 数据包处理igb_receive_skb(q_vector, skb, vlan_tag);...}...return cleaned;
}

这里主要数将 skb 摘下来，对 skb 包头一些元数据进行填充，如协议类型、时间戳等，随后就交给 igb_receive_skb 去处理：

static void igb_receive_skb(struct igb_q_vector *q_vector,struct sk_buff *skb,u16 vlan_tag)
{struct igb_adapter *adapter = q_vector->adapter;if (vlan_tag && adapter->vlgrp)vlan_gro_receive(&q_vector->napi, adapter->vlgrp,vlan_tag, skb);elsenapi_gro_receive(&q_vector->napi, skb);
}

这里主要是区分了是否 vlan 收上来的包，vlan 的包有一层独特的包头需要处理，我们之间看 napi_gro_receive 即可：

// net/core/dev.c
static gro_result_t __napi_gro_receive(struct napi_struct *napi, struct sk_buff *skb)
{struct sk_buff *p;for (p = napi->gro_list; p; p = p->next) {NAPI_GRO_CB(p)->same_flow =(p->dev == skb->dev) &&!compare_ether_header(skb_mac_header(p),skb_gro_mac_header(skb));NAPI_GRO_CB(p)->flush = 0;}return dev_gro_receive(napi, skb);
}enum gro_result dev_gro_receive(struct napi_struct *napi, struct sk_buff *skb)
{...// gro 特性处理: 将多个小包聚合成一个大包再传递给协议栈去处理// 减少传递给网络协议栈的包数，提升性能list_for_each_entry_rcu(ptype, head, list) {if (ptype->type != type || ptype->dev || !ptype->gro_receive)continue;skb_set_network_header(skb, skb_gro_offset(skb));mac_len = skb->network_header - skb->mac_header;skb->mac_len = mac_len;NAPI_GRO_CB(skb)->same_flow = 0;NAPI_GRO_CB(skb)->flush = 0;NAPI_GRO_CB(skb)->free = 0;pp = ptype->gro_receive(&napi->gro_list, skb);break;}if (pp) {struct sk_buff *nskb = *pp;*pp = nskb->next;nskb->next = NULL;// 数据包继续走 napi 流程napi_gro_complete(nskb);napi->gro_count--;}...
}

dev_gro_receive 中对小包进行了聚合，随后继续走 napi 处理流程：

static int napi_gro_complete(struct sk_buff *skb)
{// 将数据包交给协议栈处理return netif_receive_skb(skb);
}

3.2 数据包到了协议栈

接着看协议栈是如何一层一层的解包呢：

int netif_receive_skb(struct sk_buff *skb)
{// 这里设置了一个数据包的分发点，tcpdump 会监听这里的 deliver_skb 事件进行抓包list_for_each_entry_rcu(ptype, &ptype_all, list) {if (ptype->dev == null_or_orig || ptype->dev == skb->dev ||ptype->dev == orig_dev) {if (pt_prev)ret = deliver_skb(skb, pt_prev, orig_dev);pt_prev = ptype;}}// 查找对应协议注册的处理函数，放在 pt_recv 中，在 deliver_skb 中将执行它type = skb->protocol;list_for_each_entry_rcu(ptype,&ptype_base[ntohs(type) & PTYPE_HASH_MASK], list) {if (ptype->type == type && (ptype->dev == null_or_orig ||ptype->dev == skb->dev || ptype->dev == orig_dev ||ptype->dev == null_or_bond)) {if (pt_prev)ret = deliver_skb(skb, pt_prev, orig_dev);pt_prev = ptype;}}
}

netif_receive_skb，协议栈的入口函数中预留了 tcpdump 的抓包点，并跟进 skb 中的协议信息（这里是ipv4/ipv6），在 ptype_bhase 中查找对应的回调方法。随后在 deliver_skb 中，会执行对应的方法。

3.3 网络层处理

对于 IP 类型的数据包，pt_prev 中的回调方法是 ip_rcv ：

int ip_rcv(struct sk_buff *skb, struct net_device *dev, struct packet_type *pt, struct net_device *orig_dev)
{// 首先做一些包格式校验if (iph->ihl < 5 || iph->version != 4)goto inhdr_error;if (!pskb_may_pull(skb, iph->ihl*4))goto inhdr_error;iph = ip_hdr(skb);if (unlikely(ip_fast_csum((u8 *)iph, iph->ihl)))goto inhdr_error;len = ntohs(iph->tot_len);if (skb->len < len) {IP_INC_STATS_BH(dev_net(dev), IPSTATS_MIB_INTRUNCATEDPKTS);goto drop;} else if (len < (iph->ihl*4))goto inhdr_error;...// 过一下 netfilter 框架的 pre_routing 点return NF_HOOK(PF_INET, NF_INET_PRE_ROUTING, skb, dev, NULL,ip_rcv_finish);
}

ip_rcv 中一进来就会先对数据包做一些格式校验，避免非法数据包引起处理异常。接着，数据包会流经 netfilter 框架的一个 hook 点，及 PRE_ROUTING，这是数据包从网络进入主机的第一个 hook 点。在这个 hook 的点中，会执行一些钩子函数，如果数据包被放行，最终会调用 ip_rcv_finish 方法：

// net/ipv4/ip_input.c
static int ip_rcv_finish(struct sk_buff *skb)
{// 通过查路由表初始化数据包的目的地缓存项if (skb_dst(skb) == NULL) {int err = ip_route_input(skb, iph->daddr, iph->saddr, iph->tos,skb->dev);...}// 获取指向路由表的指针rt = skb_rtable(skb);// 如果路由类型是多播或者广播，就更新对应的计数器if (rt->rt_type == RTN_MULTICAST) {IP_UPD_PO_STATS_BH(dev_net(rt->u.dst.dev), IPSTATS_MIB_INMCAST,skb->len);} else if (rt->rt_type == RTN_BROADCAST)IP_UPD_PO_STATS_BH(dev_net(rt->u.dst.dev), IPSTATS_MIB_INBCAST,skb->len);// 继续处理数据包return dst_input(skb);
}

ip_rcv_finish 中涉及另一部分 skb 元数据的初始化以及多播、广播的计数更新，随后继续丢给 dst_input ：

// include/net/dst.h
static inline int dst_input(struct sk_buff *skb)
{return skb_dst(skb)->input(skb);
}// net/ipv4/ip_input.c
int ip_local_deliver(struct sk_buff *skb)
{// 重组 ip 分片if (ip_hdr(skb)->frag_off & htons(IP_MF | IP_OFFSET)) {if (ip_defrag(skb, IP_DEFRAG_LOCAL_DELIVER))return 0;}// 递交 netfilter 框架 hook 点： local_inreturn NF_HOOK(PF_INET, NF_INET_LOCAL_IN, skb, skb->dev, NULL,ip_local_deliver_finish);
}

dst_input 执行了 skb_dst 中的回调 input，其实对应的就是查路由表，决定将数据包做转发处理还是给到本机上层处理。这里我们当然看的是本机处理流程，对应的函数是 ip_local_deliver。在 ip_local_deliver 中，会先判断是否需要进行 ip 分片重组。完整的 ip 报文最终又会流经 netfilter 框架的 hook 点： LOCAL_IN。在这个 hook 的点中，会执行一些钩子函数，如果数据包被放行，那么最终会调用 ip_local_deliver_finish 方法。

static int ip_local_deliver_finish(struct sk_buff *skb){int protocol = ip_hdr(skb)->protocol;ipprot = rcu_dereference(inet_protos[protocol]);if (ipprot != NULL) {ret = ipprot->handler(skb);}
}

在这个方法中，会根据上层协议的类型，查找对应的回调函数并执行它。

3.4 传输层处理

网络层的上层自然就是传输层了，因为 tcp 的处理流程会比较复杂，为了简单理解，我们这里看 udp 的处理流程。对于 udp 来讲，它注册到 ipprot->handler 中的方法是 udp_rcv：

// net/ipv4/udp.c
int udp_rcv(struct sk_buff *skb)
{return __udp4_lib_rcv(skb, &udp_table, IPPROTO_UDP);
}int __udp4_lib_rcv(struct sk_buff *skb, struct udp_table *udptable,int proto)
{struct sock *sk;struct udphdr *uh;// 省略一些合法性校验// 查找 skb 所属的 struct socksk = __udp4_lib_lookup_skb(skb, uh->source, uh->dest, udptable);if (sk != NULL) {// 找到了int ret = udp_queue_rcv_skb(sk, skb);return 0;}// 检查 udp 校验和if (udp_lib_checksum_complete(skb))goto csum_error;// 走到这里说明没找到 sock ，发送 udp 不可达的 icmp 报文UDP_INC_STATS_BH(net, UDP_MIB_NOPORTS, proto == IPPROTO_UDPLITE);icmp_send(skb, ICMP_DEST_UNREACH, ICMP_PORT_UNREACH, 0);...
}

在 udp_rcv 里面，会根据 skb 查找对应的 struct sock 结构，如果找到了，就交给 udp_queue_rcv_skb 来处理。对于没找到的，还回复一个不可达的报文。接着看 udp_queue_rcv_skb 是如何处理的：

// net/ipv4/udp.c
int udp_queue_rcv_skb(struct sock *sk, struct sk_buff *skb)
{// 如果 udp 使用了封装，如 ipsec 协议，则调用对应协议的处理方法解封装if (up->encap_type) {/* if we're overly short, let UDP handle it */if (skb->len > sizeof(struct udphdr) &&up->encap_rcv != NULL) {int ret;ret = (*up->encap_rcv)(sk, skb);}}...// 查看 socket 是否被用户态占用if (!sock_owned_by_user(sk))rc = __udp_queue_rcv_skb(sk, skb);else if (sk_add_backlog(sk, skb)) {bh_unlock_sock(sk);goto drop;}return rc;
}

在 udp_queue_rcv_skb 中，主要是检查 socket 是否被用户态占用，即是否用户正在这个 socket 上进行系统调用。如果没有被占用，那么就将 skb 放入 socket 接收队列中；如果 socket 正在被占用，就将 skb 放在 backlog 队列中。当用户不再占用 socket 时，内核会再将 backlog 中的 skb 放到 socket 的接收队列中。总之，这里就是要把包放进 socket 的接收队列中。

4 应用层的处理

在前一篇文章《epoll 怎么就高效了》中写过，对于通过 epoll 监听的 socket，在数据包到达 socket 接收队列的时候，会遍历 socket 等待队列上的回调函数，通过 ep_poll_callback 将就绪事件通知到用户进程。对于没有通过 epoll 监听的事件，如果是那就是通过 read 或者 recvfrom 系统调用来读 socket 数据了。

recvfrom 对应的系统调用为 sys_recvfrom：

// net/socket.c
SYSCALL_DEFINE6(recvfrom, int, fd, void __user *, ubuf, size_t, size,unsigned, flags, struct sockaddr __user *, addr,int __user *, addr_len)
{...// 收包err = sock_recvmsg(sock, &msg, size, flags);// 将数据拷贝至用户空间if (err >= 0 && addr != NULL) {err2 = move_addr_to_user((struct sockaddr *)&address,msg.msg_namelen, addr, addr_len);if (err2 < 0)err = err2;}...
}

这里接着调用封装函数 sock_recvmsg 收包，收到数据后再拷贝给用户空间。

int sock_recvmsg(struct socket *sock, struct msghdr *msg,size_t size, int flags)
{ret = __sock_recvmsg(&iocb, sock, msg, size, flags);...
}static inline int __sock_recvmsg(struct kiocb *iocb, struct socket *sock,struct msghdr *msg, size_t size, int flags)
{return err ?: __sock_recvmsg_nosec(iocb, sock, msg, size, flags);
}static inline int __sock_recvmsg_nosec(struct kiocb *iocb, struct socket *sock,struct msghdr *msg, size_t size, int flags)
{...return sock->ops->recvmsg(iocb, sock, msg, size, flags);
}

这里设计一系列的封装调用，最终又是掉了 sock 上的 recvmsg 方法，对于 udp sock 来说，这个方法是 udp_recvmsg：

// net/ipv4/udp.c
int udp_recvmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,size_t len, int noblock, int flags, int *addr_len)
{...skb = __skb_recv_datagram(sk, flags | (noblock ? MSG_DONTWAIT : 0),&peeked, &err);// 设置其它出参...
}// net/core/datagram.c
struct sk_buff *__skb_recv_datagram(struct sock *sk, unsigned flags,int *peeked, int *err)
{// 循环持续尝试从接收队列中取出数据报do {// 查看 socket 接收队列中断第一个 skb，不会从队列中移除它skb = skb_peek(&sk->sk_receive_queue);if (skb) {*peeked = skb->peeked;if (flags & MSG_PEEK) {  // peek方式，增加引用计数skb->peeked = 1;atomic_inc(&skb->users);} else  // 如果不是 peek，就要从接收队列中移除__skb_unlink(skb, &sk->sk_receive_queue);}// 拿到 skb 返回了if (skb)return skb;} while (!wait_for_packet(sk, err, &timeo));return NULL;
}

__skb_recv_datagram 里终于看到了对接收队列的处理，从队列中取出 skb 然后返回。