目录
摘要
编辑
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);
else
napi_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_in
return 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 sock
sk = __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 然后返回。
5 总结
看了这么多，不免脑子已经有点乱了。有必要总结一下网卡收包大致的过程：
网卡收到数据包，DMA 拷贝至 RingBuf，发出硬中断cpu 执行网卡注册的硬中断处理函数，将数据挂到 softnet_data 的 poll_list 上，发出软中断ksoftirq/x 处理软中断，将数据包从 RingBuf 中取出，交给协议栈协议栈层层处理，经网络层交给传输层，数据包被放到 socket 的接收队列中应用层调用 recvfrom 从接收队列中取数据
可以看出收一个网络包的处理过程很是繁杂，为了优化性能，这里又涉及硬中断到多个cpu的负载均衡，进协议栈前网卡 gro 特性做的小包聚合，以及文中没有写出来的收到多个包才会聚合发出一个硬中断。革命尚未成功，同志们仍需努力呀！
好文推荐
发表评论