深入理解Linux网络随笔（五）：深度理解本机网络I/O

分析本机网络I/O部分源码需要知道本机I/O是什么？扮演什么角色？

本机网络 I/O（Local Network I/O）是指发生在同一台设备上的网络输入/输出操作，数据在本机内部流转，而不经过外部网络。主要包括以下几种情况：

（1）回环通信

通过 127.0.0.1（IPv4）或 ::1（IPv6） 进行的网络通信，发送到 回环接口 lo，数据不会经过网卡，而是直接在内核中处理。

（2）主机内部不同进程的网络通信

两个进程使用本机 IP（非 127.0.0.1）进行通信。

（3）通过 TUN/TAP 设备的虚拟网络通信

VPN、容器、虚拟网络设备 等使用 TUN/TAP 设备进行本机 I/O，关于本机网络I/O主要分为两部分内容：本机发送过程和本机接收过程。

本机发送过程

前几篇文章中已经分析了网络层的入口函数是ip_queue_xmit，在网络层会进行路由相关工作。

int __ip_queue_xmit(struct sock *sk, struct sk_buff *skb, struct flowi *fl,__u8 tos)
{struct inet_sock *inet = inet_sk(sk);struct net *net = sock_net(sk);struct ip_options_rcu *inet_opt;struct flowi4 *fl4;struct rtable *rt;struct iphdr *iph;int res;//检查socket是否有缓存的路由表/* Make sure we can route this packet. */rt = (struct rtable *)__sk_dst_check(sk, 0);if (!rt) {__be32 daddr;/* Use correct destination address if we have options. */daddr = inet->inet_daddr;if (inet_opt && inet_opt->opt.srr)daddr = inet_opt->opt.faddr;/* If this fails, retransmit mechanism of transport layer will* keep trying until route appears or the connection times* itself out.*///查找路由rt = ip_route_output_ports(net, fl4, sk,daddr, inet->inet_saddr,inet->inet_dport,inet->inet_sport,sk->sk_protocol,RT_CONN_FLAGS_TOS(sk, tos),sk->sk_bound_dev_if);if (IS_ERR(rt))goto no_route;//设置路由信息到skbsk_setup_caps(sk, &rt->dst);}skb_dst_set_noref(skb, &rt->dst);
}

调用逻辑ip_route_output_ports-->ip_route_output_flow-->__ip_route_output_key-->ip_route_output_key_hash-->ip_route_output_key_hash_rcu，在ip_route_output_key_hash_rcu函数中完成路由查找。

struct rtable *ip_route_output_key_hash_rcu(struct net *net, struct flowi4 *fl4,struct fib_result *res,const struct sk_buff *skb)
{struct net_device *dev_out = NULL;//网络设备loint orig_oif = fl4->flowi4_oif;//原始输出接口索引unsigned int flags = 0;//路由标志位struct rtable *rth;//路由表条目int err;// 执行查找转发信息库（FIB），获取路由err = fib_lookup(net, fl4, res, 0);if (err) {res->fi = NULL;res->table = NULL;if (fl4->flowi4_oif &&(ipv4_is_multicast(fl4->daddr) || !fl4->flowi4_l3mdev)) {if (fl4->saddr == 0)fl4->saddr = inet_select_addr(dev_out, 0,RT_SCOPE_LINK);res->type = RTN_UNICAST;//单播路由goto make_route;}rth = ERR_PTR(err);goto out;}//路由类型是本地路由if (res->type == RTN_LOCAL) {if (!fl4->saddr) {if (res->fi->fib_prefsrc)//存在首选源地址fl4->saddr = res->fi->fib_prefsrc;else//否则选择目标地址fl4->saddr = fl4->daddr;}//根据路由结果设备确定L3主设备，默认回环设备dev_out = l3mdev_master_dev_rcu(FIB_RES_DEV(*res)) ? :net->loopback_dev;//FIB输出接口orig_oif = FIB_RES_OIF(*res);//设置输出接口fl4->flowi4_oif = dev_out->ifindex;flags |= RTCF_LOCAL;//设置本地标志goto make_route;}fib_select_path(net, res, fl4, skb);dev_out = FIB_RES_DEV(*res);// 获取路由的输出设备
}

针对传入的路由类型进行操作，RTN_LOCAL类型是对本地local表展开查询，并设置了net->loopback_dev，所以不论是本机IP还是127.0.0.1回环地址都会被添加到 local 路由表，查询完成之后调用fib_lookup，fib_lookup会先检查local表，只有未在local表找到匹配项时才会继续查找main表。这种设计优化了本机通信的性能，因为本机通信的数据包只需要在local表中查找即可。

static inline int fib_lookup(struct net *net, const struct flowi4 *flp,struct fib_result *res, unsigned int flags)
{struct fib_table *tb;int err = -ENETUNREACH;  // 默认返回 "网络不可达" 错误// 加读锁，保护 RCU 访问的 FIB 结构rcu_read_lock();// 获取主路由表 RT_TABLE_MAIN（ID = 254）tb = fib_get_table(net, RT_TABLE_MAIN);if (tb)// 在主路由表中查找匹配的路由项err = fib_table_lookup(tb, flp, res, flags | FIB_LOOKUP_NOREF);// 如果查找返回 -EAGAIN，则转换为 -ENETUNREACHif (err == -EAGAIN)err = -ENETUNREACH;// 释放 RCU 读锁rcu_read_unlock();return err;  // 返回查找结果
}

通过ip route list table local查看本地路由表，但是一般我们只看到了local表，为什么？因为在fib_lookup() 路由查找函数，它会依次检查多个路由表，但当查找到 local 表时，就会直接返回，不再继续查找 main 表，只有当 local 表未匹配时，才会去 main 表查询。如下图，可以看到直接走 lo 设备，根本不会去 main 表查找。

数据包经过网络层到达邻居子系统再到网络设备子系统，前文也分析了网络设备子系统的入口函数是dev_queue_xmit。对于普通网卡，数据包进入__dev_xmit_skb入队，当q->enqueue为空，说明当前是lo回环设备，调用dev_hard_start_xmit进行入队。

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;bool again = false;if (q->enqueue) {rc = __dev_xmit_skb(skb, q, dev, txq);goto out;
}if (dev->flags & IFF_UP) {...skb = dev_hard_start_xmit(skb, dev, txq, &rc);
}}

通过 dev->netdev_ops 获取到该设备的操作集合 netdev_ops，包含了设备相关操作的指针，然后调用__netdev_start_xmit实际执行数据包的发送操作。

static inline netdev_tx_t netdev_start_xmit(struct sk_buff *skb, struct net_device *dev,struct netdev_queue *txq, bool more)
{const struct net_device_ops *ops = dev->netdev_ops;netdev_tx_t rc;rc = __netdev_start_xmit(ops, skb, dev, more);if (rc == NETDEV_TX_OK)txq_trans_update(txq);return rc;
}

static const struct net_device_ops loopback_ops = {.ndo_init        = loopback_dev_init,//初始化.ndo_start_xmit  = loopback_xmit,//发送数据.ndo_get_stats64 = loopback_get_stats64,//获取回环接口的64位信息.ndo_set_mac_address = eth_mac_addr,//设置网络设备的 MAC 地址
};

调用skb_orphan将数据包与原始 socket 之间的联系，将其状态设置为“孤立”，因为lo回环设备用于数据包的自我发送和接收。当数据包从一个应用程序发送到环回设备时，它会返回到发送应用程序。这个过程不会经过网络堆栈的复杂路由和传输层处理，而是直接通过环回接口返回。所以不希望将数据包与某个特定的应用层套接字直接绑定，仅仅在设备间完成传输。接着调用__netif_rx发送数据包。

static netdev_tx_t loopback_xmit(struct sk_buff *skb,struct net_device *dev)S
{int len;//计算数据包发送时间戳skb_tx_timestamp(skb);/* do not fool net_timestamp_check() with various clock bases *///清除skb_clear_tstamp(skb);//剥离掉和原socket联系skb_orphan(skb);/* Before queueing this packet to __netif_rx(),* make sure dst is refcounted.*/skb_dst_force(skb);skb->protocol = eth_type_trans(skb, dev);len = skb->len;//数据包交付if (likely(__netif_rx(skb) == NET_RX_SUCCESS))dev_lstats_add(dev, len);return NETDEV_TX_OK;
}

函数调用逻辑__netif_rx-->netif_rx_internal-->enqueue_to_backlog。该函数将skb添加在等待队列尾部，主要完成三件事：

1、使用per_cpu宏访问每个CPU核心对应的softnet_data数据结构（softnet_data保存每个CPU核心的网络接收队列等，避免CPU之间竞争资源）

2、__skb_queue_tail将skb添加到等待队列尾部

3、napi_schedule_rpsNAPI机制处理接收的数据包

static int enqueue_to_backlog(struct sk_buff *skb, int cpu,unsigned int *qtail)
{enum skb_drop_reason reason;struct softnet_data *sd;unsigned long flags;unsigned int qlen;//丢包原因：未指定reason = SKB_DROP_REASON_NOT_SPECIFIED;//per_cpu宏访问每个CPU核心对应的softnet_data数据结构sd = &per_cpu(softnet_data, cpu);//获取锁保存当前中断状态rps_lock_irqsave(sd, &flags);if (!netif_running(skb->dev))goto drop;//获取接收队列长度qlen = skb_queue_len(&sd->input_pkt_queue);//检查队列长度是否小于等于网络设备的最大容纳的队列长度，是否到达了流量限制if (qlen <= READ_ONCE(netdev_max_backlog) && !skb_flow_limit(skb, qlen)) {if (qlen) {
enqueue:
//将skb加入等待队列尾部__skb_queue_tail(&sd->input_pkt_queue, skb);//更新队尾指针并保存队列状态input_queue_tail_incr_save(sd, qtail);//释放锁恢复中断rps_unlock_irq_restore(sd, &flags);//入队成功return NET_RX_SUCCESS;}/* Schedule NAPI for backlog device* We can use non atomic operation since we own the queue lock*///NAPI处理if (!__test_and_set_bit(NAPI_STATE_SCHED, &sd->backlog.state))napi_schedule_rps(sd);goto enqueue;}reason = SKB_DROP_REASON_CPU_BACKLOG;......
}

这个函数分为两种情况，分别是将开启RPS和不开启RPS，开启RPS机制时，当接收到网络数据包，RPS 会决定是否将数据包分配给其他 CPU 进行处理。如果是，当前 CPU 会把任务添加到目标 CPU 的 rps_ipi_list 中，并触发软中断。如果没有开启RPS那么就是NAPI轮询调度数据包，调用__napi_schedule_irqoff函数。

static int napi_schedule_rps(struct softnet_data *sd)
{struct softnet_data *mysd = this_cpu_ptr(&softnet_data);#ifdef CONFIG_RPSif (sd != mysd) {sd->rps_ipi_next = mysd->rps_ipi_list;mysd->rps_ipi_list = sd;__raise_softirq_irqoff(NET_RX_SOFTIRQ);return 1;}#endif /* CONFIG_RPS */__napi_schedule_irqoff(&mysd->backlog);return 0;
}

__napi_schedule_irqoff主要时调度NAPI任务，处理网络接收的队列，根据开启的内核配置分两种调度场景，开启CONFIG_PREEMPT_RT采用实时调度策略，调用__napi_schedule，非实时调度根据当前 CPU 的软中断队列和网络数据结构来进行调度，调用函数____napi_schedule。

void __napi_schedule_irqoff(struct napi_struct *n)
{//非实时调度if (!IS_ENABLED(CONFIG_PREEMPT_RT))//NAPI调度____napi_schedule(this_cpu_ptr(&softnet_data), n);//实时调度else__napi_schedule(n);
}
EXPORT_SYMBOL(__napi_schedule_irqoff);

input_pkt_queue 是 softnet_data 结构中的一个队列，用于存储接收到的网络数据包（skb），____napi_schedule主要完成两件事：

1、将发送的skb添加到softnet_data的input_pkt_queue队列；

2、触发软中断

static inline void ____napi_schedule(struct softnet_data *sd,struct napi_struct *napi)
{struct task_struct *thread;lockdep_assert_irqs_disabled();
......//将napi添加到当前CPU的softnet_data结构体的poll_list链表尾部list_add_tail(&napi->poll_list, &sd->poll_list);//触发软中断，类型NET_RX_SOFTIRQ__raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

本机接收过程

本机I/O接收过程主要是驱动部分存在不同，在初始化的时候会设置数据包等待队列的回调函数sd->backlog.poll = process_backlog，核心是skb_queue_splice_tail_init函数的调用，将 sd->input_pkt_queue 中的所有元素（即网络数据包）移动到 sd->process_queue 队列的尾部。

• input_pkt_queue队列：接收的网络数据包队列
• process_queue队列：将要被处理的数据包队列

static int process_backlog(struct napi_struct *napi, int quota)
{struct softnet_data *sd = container_of(napi, struct softnet_data, backlog);bool again = true;int work = 0;/* Check if we have pending ipi, its better to send them now,* not waiting net_rx_action() end.*///是否存在等待处理的rps中断if (sd_has_rps_ipi_waiting(sd)) {local_irq_disable();net_rps_action_and_irq_enable(sd);}//读取设备的最大接受数据包的数量napi->weight = READ_ONCE(dev_rx_weight);while (again) {struct sk_buff *skb;//从process_queue队列取出skbwhile ((skb = __skb_dequeue(&sd->process_queue))) {rcu_read_lock();//发送数据包__netif_receive_skb(skb);rcu_read_unlock();//更新数据包，包已处理input_queue_head_incr(sd);if (++work >= quota)return work;}rps_lock_irq_disable(sd);//input_pkt_queue 等待队列空if (skb_queue_empty(&sd->input_pkt_queue)) {/** Inline a custom version of __napi_complete().* only current cpu owns and manipulates this napi,* and NAPI_STATE_SCHED is the only possible flag set* on backlog.* We can use a plain write instead of clear_bit(),* and we dont need an smp_mb() memory barrier.*/napi->state = 0;again = false;} else {//队列合并skb_queue_splice_tail_init(&sd->input_pkt_queue,&sd->process_queue);}rps_unlock_irq_enable(sd);}return work;
}

总结

本机网络I/O并不会经过网卡，Linux 内核在进行路由查找时，优先查询 RT_TABLE_LOCAL 路由表，具体表现如下：

• 如果目标 IP 是本机地址（如 127.0.0.1 或者主机上任意 IP 地址），查找 RT_TABLE_LOCAL 并匹配 RTN_LOCAL 路由类型，数据包的 下一跳（next hop） 指向 loopback_dev（即回环设备）。
• 如果未命中 RT_TABLE_LOCAL，才会查 main 路由表，决定是否通过真实网卡发送。