深入理解Linux网络（八）：内核如何发送网络包

一、总览

⽤户数据被拷⻉到内核态，然后经过协议栈处理后进⼊到了 RingBuffer 中。随后⽹卡驱动真正将数据发送了出去。
当发送完成，硬中断来通知 CPU 触发清理 RingBuffer 的代码。
调用顺序如下：

应用层：send()
系统调用：
SYSCALL_DEFINE6(sendto, int, fd, ...)sock_sendmsg(sock, &msg, len)__sock_sendmsg_nosec()sock->ops->sendmsg(iocb, sock, msg, size)
协议层：
inet_sendmsg()sk-sk_prot->sendmsg(iocb, sk, msg, size)传输层：tcp_sendmsg()tcp_transmit_skb()icsk->icsk_af_ops->queue_xmit(skb)网络层：ip_queue_xmit(struct sk_buff *skb, struct flowi *fi)ip_local_out(skb)ip_finish_output2(struct sk_buff *skb)dst_neigh_output(dst, neigh, skb)neigh_hh_output(hh, skb)dev_queue_xmit(skb)
网络设备子系统：net/core/dev.c
dev_queue_xmit(struct sk_buff *skb)__dev_xmit_skb(skb, q, dev, txq)dev_hard_start_xmit(...)ops->ndo_start_xxmit(skb, dev)
驱动：（不同硬件驱动不一样）
igb_xmit_frame(...)igb_xmit_frame_ring(skb, ...)igb_tx_map(tx_ring, first, hdr_len)
硬件：射频发出

发送完毕后，需要释放缓存队列等内存，调用顺序如下：

硬件：硬中断（由驱动调用）
igb_msix_ring(int irq, void *data)napi_schedule(&q_vector->napi)____napi_schedule(...)__raise_softirq_irqoff(NET_RX_SOFTIRQ)
软中断：
net_rx_action(struct softirq_action *h)n->poll(n, weight)
驱动：
igb_poll(struct napi_struct *napi, int budget)igb_clean_tx_irq(q_vecor)igb_clean_tx_irq(struct igb_q_vector *q_vector)//释放 skbdev_kfree_skb_any(tx_buffer->skb);//清楚 tx_buffretx_buffer->skb = NULL;dma_unmap_len_set(tx_buffer, len, 0);//清理 DMA 区域while (tx_desc != eop_desc){...}

可以看到硬中断最终触发的软中断是 NET_RX_SOFTIRQ，⽽并不是 NET_TX_SOFTIRQ！！！
因此，在服务器上查看 /proc/softirqs，NET_RX 要⽐ NET_TX ⼤的多。

二、网卡启动准备

⽹卡⼀般都是⽀持多队列的。每⼀个队列上都是由⼀个 RingBuffer 表示的，开启了多队列以后的的⽹卡就会对应有多个 RingBuffer。

⽹卡在启动时最重要的任务之⼀就是分配和初始化 RingBuffer。
igb网卡的分配在 __igb_open() 中：

//file: drivers/net/ethernet/intel/igb/igb_main.c
static int __igb_open(struct net_device *netdev, bool resuming)
{struct igb_adapter *adapter = netdev_priv(netdev);//分配传输描述符数组err = igb_setup_all_tx_resources(adapter);//分配接收描述符数组err = igb_setup_all_rx_resources(adapter);//开启全部队列netif_tx_start_all_queues(netdev);
}

在上⾯ __igb_open 函数调⽤ igb_setup_all_tx_resources 分配所有的传输 RingBuffer, 调⽤ igb_setup_all_rx_resources 创建所有的接收 RingBuffer。

//file: drivers/net/ethernet/intel/igb/igb_main.c
static int igb_setup_all_tx_resources(struct igb_adapter
*adapter)
{//有⼏个队列就构造⼏个 RingBufferfor (i = 0; i < adapter->num_tx_queues; i++) {igb_setup_tx_resources(adapter->tx_ring[i]);}
}

真正的 RingBuffer 构造过程是在 igb_setup_tx_resources 中完成的。

//file: drivers/net/ethernet/intel/igb/igb_main.c
int igb_setup_tx_resources(struct igb_ring *tx_ring)
{//1.申请 igb_tx_buffer 数组内存size = sizeof(struct igb_tx_buffer) * tx_ring->count;tx_ring->tx_buffer_info = vzalloc(size);//2.申请 e1000_adv_tx_desc DMA 数组内存tx_ring->size = tx_ring->count * sizeof(union e1000_adv_tx_desc);tx_ring->size = ALIGN(tx_ring->size, 4096);tx_ring->desc = dma_alloc_coherent(dev, tx_ring->size,&tx_ring->dma, GFP_KERNEL);//3.初始化队列成员tx_ring->next_to_use = 0;tx_ring->next_to_clean = 0;
}

从上述源码可以看到，实际上⼀个 RingBuffer 的内部不仅仅是⼀个环形队列数组，⽽是有两个。

1. igb_tx_buffer 数组：这个数组是内核使⽤的，通过 vzalloc 申请的。
2. e1000_adv_tx_desc 数组：这个数组是⽹卡硬件使⽤的，硬件是可以通过 DMA 直接访问这块内存，通过 dma_alloc_coherent 分配。

这个时候它们之间还没有啥联系。将来在发送的时候，这两个环形数组中相同位置的指针将都将指向同⼀个 skb。这样，内核和硬件就能共同访问同样的数据了，内核往 skb ⾥写数据，⽹卡硬件负责发送。

最后调⽤ netif_tx_start_all_queues 开启队列。另外，对于硬中断的处理函数 igb_msix_ring 其实也是在 __igb_open 中注册的。

三、ACCEPT 创建新 SOCKET

当 accept 之后，进程会创建⼀个新的 socket 出来，然后把它放到当前进程的打开⽂件列表中，专⻔⽤于和对应的客户端通信。
假设服务器进程通过 accept 和客户端建⽴了两条连接，我们来简单看⼀下这两条连接和进程的关联关系。

其中代表⼀条连接的 socket 内核对象更为具体⼀点的结构图如下：

四、开始发送数据

send 系统调⽤实现

send 系统调⽤的源码位于⽂件 net/socket.c 中，在这个系统调⽤使⽤了 sendto 系统调⽤。主要做了两件简单的事情：
第⼀是在内核中把真正的 socket 找出来，在这个对象⾥记录着各种协议栈的函数地址。
第⼆是构造⼀个 struct msghdr 对象，把⽤户传⼊的数据，⽐如 buffer地址、数据⻓度啥的，统统都装进去。

剩下的事情就交给下⼀层，协议栈⾥的函数 inet_sendmsg 了，其中 inet_sendmsg 函数的地址是通过 socket 内核对象⾥的 ops 成员找到的。

源码如下：

//file: net/socket.c
SYSCALL_DEFINE4(send, int, fd, void __user *, buff, size_t, len, unsigned int, flags)
{return sys_sendto(fd, buff, len, flags, NULL, 0);
}
SYSCALL_DEFINE6(......)
{//1.根据 fd 查找到 socketsock = sockfd_lookup_light(fd, &err, &fput_needed);//2.构造 msghdrstruct msghdr msg;struct iovec iov;iov.iov_base = buff;iov.iov_len = len;msg.msg_iovlen = 1;msg.msg_iov = &iov;msg.msg_flags = flags;......//3.发送数据sock_sendmsg(sock, &msg, len);
}

从源码可以看到，我们在⽤户态使⽤的 send 函数和 sendto 函数其实都是 sendto 系统调⽤实现的。send 只是为了⽅便，封装出来的⼀个更易于调⽤的⽅式⽽已。
在 sendto 系统调⽤⾥，⾸先根据⽤户传进来的 socket 句柄号来查找真正的 socket 内核对象。接着把⽤户请求的 buff、len、flag 等参数都统统打包到⼀个 struct msghdr 对象中。
接着调⽤了 sock_sendmsg => __sock_sendmsg ==> __sock_sendmsg_nosec。
在__sock_sendmsg_nosec 中，调⽤将会由系统调⽤进⼊到协议栈，我们来看它的码。

static inline int __sock_sendmsg_nosec(...)
{......return sock->ops->sendmsg(iocb, sock, msg, size);
}

通过 socket 内核对象结构，我们可以看到，这⾥调⽤的是 sock->ops->sendmsg 实际执⾏的是 inet_sendmsg。这个函数是 AF_INET 协议族提供的通⽤发送函数。

传输层处理

传输层拷贝

在进⼊到协议栈 inet_sendmsg 以后，内核接着会找到 socket 上的具体协议发送函数。对于 TCP 协议来说，那就是 tcp_sendmsg（同样也是通过 socket 内核对象找到的）。
在这个函数中，内核会申请⼀个内核态的 skb 内存，将⽤户待发送的数据拷⻉进去。注意这个时候不⼀定会真正开始发送，如果没有达到发送条件的话很可能这次调⽤直接就返回了。
如图：

//file: net/ipv4/af_inet.c
int inet_sendmsg(......)
{......return sk->sk_prot->sendmsg(iocb, sk, msg, size);
}

在这个函数中会调⽤到具体协议的发送函数。同样参考第三节⾥的 socket 内核对象结构图，
我们看到对于 TCP 协议下的 socket 来说，来说 sk->sk_prot->sendmsg 指向的是 tcp_sendmsg（对于 UDP 来说是 udp_sendmsg）。
tcp_sendmsg 这个函数⽐较⻓，我们分多次来看它。 先看这⼀段

//file: net/ipv4/tcp.c
int tcp_sendmsg(...)
{while(...){while(...){//获取发送队列skb = tcp_write_queue_tail(sk);//申请skb 并拷⻉......}}
}//file: include/net/tcp.hstatic
inline struct sk_buff *tcp_write_queue_tail(const struct sock *sk)
{return skb_peek_tail(&sk->sk_write_queue);
}

理解对 socket 调⽤ tcp_write_queue_tail 是理解发送的前提。如上所示，这个函数是在获取 socket 发送队列中的最后⼀个 skb。 skb 是 struct sk_buff 对象的简称，⽤户的发送队列就是该对象组成的⼀个链表。

我们再接着看 tcp_sendmsg 的其它部分。

//file: net/ipv4/tcp.c
int tcp_sendmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg, size_t size){//获取⽤户传递过来的数据和标志iov = msg->msg_iov; //⽤户数据地址iovlen = msg->msg_iovlen; //数据块数为1flags = msg->msg_flags; //各种标志 //遍历⽤户层的数据块while (--iovlen >= 0) { //待发送数据块的地址 unsigned char __user *from = iov->iov_base; while (seglen > 0) { //需要申请新的 skbif (copy <= 0) { //申请 skb，并添加到发送队列的尾部skb = sk_stream_alloc_skb(sk, select_size(sk, sg), sk->sk_allocation); //把 skb 挂到socket的发送队列上skb_entail(sk, skb); } // skb 中有⾜够的空间if (skb_availroom(skb) > 0) { //拷⻉⽤户空间的数据到内核空间，同时计算校验和//from是⽤户空间的数据地址skb_add_data_nocache(sk, skb, from, copy); } ......

这个函数⽐较⻓，不过其实逻辑并不复杂。其中 msg->msg_iov 存储的是⽤户态内存的要发送的数据的 buffer。接下来在内核态申请内核内存，⽐如 skb，并把⽤户内存⾥的数据拷⻉到内核态内存中。这就会涉及到⼀次或者⼏次内存拷⻉的开销。

⾄于内核什么时候真正把 skb 发送出去。在 tcp_sendmsg 中会进⾏⼀些判断。

//file: net/ipv4/tcp.c
int tcp_sendmsg(...){while(...){ while(...){ //申请内核内存并进⾏拷⻉//发送判断if(forced_push(tp)) {tcp_mark_push(tp, skb); __tcp_push_pending_frames(sk, mss_now, TCP_NAGLE_PUSH);}else if ( skb == tcp_send_head(sk))tcp_push_one(sk,mss_now); } continue;}}
}

只有满⾜ forced_push(tp) 或者 skb == tcp_send_head(sk) 成⽴的时候，内核才会真正启动发送数据包。其中 forced_push(tp) 判断的是未发送的数据数据是否已经超过最⼤窗⼝的⼀半了。
条件都不满⾜的话，这次的⽤户要发送的数据只是拷⻉到内核就算完事了！

传输层发送

假设现在内核发送条件已经满⾜了，我们再来跟踪⼀下实际的发送过程。 对于上⼩节函数中，当满⾜真正发送条件的时候，⽆论调⽤的是 __tcp_push_pending_frames 还是 tcp_push_one 最终都实际会执⾏到 tcp_write_xmit。
所以我们直接从 tcp_write_xmit 看起，这个函数处理了传输层的拥塞控制、滑动窗⼝相关的⼯作。满⾜窗⼝要求的时候，设置⼀下 TCP 头然后将 skb 传到更低的⽹络层进⾏处理。

//file: net/ipv4/tcp_output.c
static bool tcp_write_xmit(struct sock *sk, unsigned int mss_now,int nonagle, int push_one, gfp_t gfp){//循环获取待发送 skbwhile ((skb = tcp_send_head(sk))) {//滑动窗⼝相关 cwnd_quota =tcp_cwnd_test(tp, skb);tcp_snd_wnd_test(tp, skb, mss_now); tcp_mss_split_point(...);tso_fragment(sk, skb, ...); ......//真正开启发送tcp_transmit_skb(sk, skb, 1, gfp); }
}

可以看到我们之前在⽹络协议⾥学的滑动窗⼝、拥塞控制就是在这个函数中完成的，这部分就不过多展开了，感兴趣同学⾃⼰找这段源码来读。我们今天只看发送主过程，那就⾛到了 tcp_transmit_skb。

//file: net/ipv4/tcp_output.c
static int tcp_transmit_skb(struct sock *sk, struct sk_buff *skb, int clone_it, gfp_t gfp_mask){//1.克隆新 skb 出来if (likely(clone_it)) { skb = skb_clone(skb, gfp_mask); ...... }//2.封装 TCP 头th = tcp_hdr(skb);th->source = inet->inet_sport;th->dest = inet->inet_dport;th->window = ...;th->urg = ...; ......//3.调⽤⽹络层发送接⼝err = icsk->icsk_af_ops->queue_xmit(skb, &inet->cork.fl);}

首先要克隆⼀个新的 skb，这⾥重点说下为什么要复制⼀个 skb 出来。
因为 skb 后续在调⽤⽹络层，最后到达⽹卡发送完成的时候，这个 skb 会被释放掉。
⽽我们知道 TCP 协议是⽀持丢失重传的，在收到对⽅的 ACK 之前，这个 skb 不能被删除。
所以内核的做法就是每次调⽤⽹卡发送的时候，实际上传递出去的是 skb 的⼀个拷⻉。等收到 ACK 再真正删除。
其次就是修改 skb 中的 TCP header，根据实际情况把 TCP 头设置好。
skb 内部其实包含了⽹络协议中所有的 header。
在设置 TCP 头的时候，只是把指针指向 skb 的合适位置。
后⾯再设置 IP 头的时候，在把指针挪⼀挪就⾏，避免频繁的内存申请和拷⻉，效率很⾼。

tcp_transmit_skb 是发送数据位于传输层的最后⼀步，接下来就可以进⼊到⽹络层进⾏下⼀层的操作了。调⽤了⽹络层提供的发送接⼝icsk->icsk_af_ops->queue_xmit()。
在下⾯的这个源码中，我们的知道了 queue_xmit 其实指向的是 ip_queue_xmit 函数。

//file: net/ipv4/tcp_ipv4.c
const struct inet_connection_sock_af_ops ipv4_specific = {.queue_xmit = ip_queue_xmit,.send_check = tcp_v4_send_check,...
}

传输层的工作完成了。 数据离开了传输层，将会进⼊网络层。

网络层发送原理

Linux 内核⽹络层的发送的实现位于 net/ipv4/ip_output.c 这个⽂件。传输层调⽤到的 ip_queue_xmit 也在这⾥。（从⽂件名上也能看出来进⼊到 IP 层了，源⽂件名已经从tcp_xxx 变成了 ip_xxx。）
在⽹络层⾥主要处理路由项查找、IP 头设置、netfilter 过滤、skb 切分（⼤于 MTU 的话）等⼏项⼯作，处理完这些⼯作后会交给更下层的邻居⼦系统来处理。

我们来看⽹络层⼊⼝函数 ip_queue_xmit 的源码：

//file: net/ipv4/ip_output.c
int ip_queue_xmit(struct sk_buff *skb, struct flowi *fl)
{//检查 socket 中是否有缓存的路由表rt = (struct rtable *)__sk_dst_check(sk, 0);if (rt == NULL) {//没有缓存则展开查找路由项， 并缓存到 socket 中rt = ip_route_output_ports(...);sk_setup_caps(sk, &rt->dst);}//为 skb 设置路由表skb_dst_set_noref(skb, &rt->dst);//设置 IP headeriph = ip_hdr(skb);iph->protocol = sk->sk_protocol;iph->ttl = ip_select_ttl(inet, &rt->dst);iph->frag_off = ...;//发送ip_local_out(skb);
}

ip_queue_xmit 已经到了⽹络层，在这个函数⾥我们看到了⽹络层相关的功能路由项查找，如果找到了则设置到 skb 上（没有路由的话就直接报错返回了）。
在 Linux 上通过 route 命令可以看到你本机的路由配置。

在路由表中，可以查到某个⽬的⽹络应该通过哪个 Iface（⽹卡），哪个 Gateway（⽹卡）发送出去。查找出来以后缓存到 socket 上，下次再发送数据就不⽤查了。
接着把路由表地址也放到 skb ⾥去。

//file: include/linux/skbuff.h
struct sk_buff {//保存了⼀些路由相关信息unsigned long _skb_refdst;
}

接下来就是定位到 skb ⾥的 IP 头的位置上，然后开始按照协议规范设置 IP header。

再通过 ip_local_out 进⼊到下⼀步的处理。

//file: net/ipv4/ip_output.c 
int ip_local_out(struct sk_buff *skb)
{//执⾏ netfilter 过滤err = __ip_local_out(skb);//开始发送数据if (likely(err == 1))err = dst_output(skb);......

在 ip_local_out => __ip_local_out => nf_hook 会执⾏ netfilter 过滤。如果你使⽤ iptables 配置了⼀些规则，那么这⾥将检测是否命中规则。
如果你设置了⾮常复杂的 netfilter 规则，在这个函数这⾥将会导致你的进程 CPU 开销会极⼤增加。
还是不多展开说，继续只聊和发送有关的过程 dst_output。

//file: include/net/dst.h
static inline int dst_output(struct sk_buff *skb)
{return skb_dst(skb)->output(skb);
}

此函数找到到这个 skb 的路由表（dst 条⽬） ，然后调⽤路由表的 output ⽅法。这⼜是⼀个函数指针，指向的是 ip_output ⽅法。

//file: net/ipv4/ip_output.c
int ip_output(struct sk_buff *skb)
{//统计.....//再次交给 netfilter，完毕后回调 ip_finish_outputreturn NF_HOOK_COND(NFPROTO_IPV4, NF_INET_POST_ROUTING, skb, NULL, dev,ip_finish_output, !(IPCB(skb)->flags & IPSKB_REROUTED));
}

在 ip_output 中进⾏⼀些简单的，统计⼯作，再次执⾏ netfilter 过滤。过滤通过之后回调 ip_finish_output。

//file: net/ipv4/ip_output.c
static int ip_finish_output(struct sk_buff *skb)
{//⼤于 mtu 的话就要进⾏分⽚了if (skb->len > ip_skb_dst_mtu(skb) && !skb_is_gso(skb))return ip_fragment(skb, ip_finish_output2);elsereturn ip_finish_output2(skb);
}

在 ip_finish_output 中我们看到，如果数据⼤于 MTU 的话，是会执⾏分⽚的。

注意：实际 MTU ⼤⼩确定依赖 MTU 发现，以太⽹帧为 1500 字节。之前 QQ 团队在早期的时候，会尽量控制⾃⼰数据包尺⼨⼩于 MTU，通过这种⽅式来优化⽹络性能。
因为分⽚会带来两个问题：
1、需要进⾏额外的切分处理，有额外性能开销。
2、只要⼀个分⽚丢失，整个包都得重传。所以避免分⽚既杜绝了分⽚开销，也⼤⼤降低了重传率。

在 ip_finish_output2 中，终于发送过程会进⼊到下⼀层，邻居⼦系统中。

//file: net/ipv4/ip_output.c
static inline int ip_finish_output2(struct sk_buff *skb)
{//根据下⼀跳 IP 地址查找邻居项，找不到就创建⼀个nexthop = (__force u32) rt_nexthop(rt, ip_hdr(skb)->daddr); neigh = __ipv4_neigh_lookup_noref(dev, nexthop);if (unlikely(!neigh))neigh = __neigh_create(&arp_tbl, &nexthop, dev, false);//继续向下层传递int res = dst_neigh_output(dst, neigh, skb);
}

邻居⼦系统

邻居⼦系统是位于⽹络层和数据链路层中间的⼀个系统，其作⽤是对⽹络层提供⼀个封装，让⽹络层不必关⼼下层的地址信息，让下层来决定发送到哪个 MAC 地址。
⽽且这个邻居⼦系统并不位于协议栈 net/ipv4/ ⽬录内，⽽是位于 net/core/neighbour.c。
因为⽆论是对于 IPv4 还是 IPv6 ，都需要使⽤该模块。

在邻居子系统中主要是查找或者创建邻居项，在创造邻居项时，有可能发出实际的 arp 请求。然后封装一下 MAC 头，将发送过程传递到更下层的网咯设备子系统。如下所示：

ip_finish_output2 源码中调⽤了 __ipv4_neigh_lookup_noref。它是在 arp 缓存中进⾏查找，其第⼆个参数传⼊的是路由下⼀跳 IP 信息。

//file: include/net/arp.h
extern struct neigh_table arp_tbl;
static inline struct neighbour *__ipv4_neigh_lookup_noref(struct net_device *dev, u32 key)
{struct neigh_hash_table *nht = rcu_dereference_bh(arp_tbl.nht);//计算 hash 值，加速查找hash_val = arp_hashfn(......);for (n = rcu_dereference_bh(nht->hash_buckets[hash_val]);n != NULL;n = rcu_dereference_bh(n->next)) {if (n->dev == dev && *(u32 *)n->primary_key == key)return n;}
}

如果查找不到，则调⽤ __neigh_create 创建⼀个邻居。

//file: net/core/neighbour.c
struct neighbour *__neigh_create(......)
{//申请邻居表项struct neighbour *n1, *rc, *n = neigh_alloc(tbl, dev);//构造赋值memcpy(n->primary_key, pkey, key_len);n->dev = dev;n->parms->neigh_setup(n);//最后添加到邻居 hashtable 中rcu_assign_pointer(nht->hash_buckets[hash_val], n);......

有了邻居项以后，此时仍然还不具备发送 IP 报⽂的能⼒，因为⽬的 MAC 地址还未获取。
调⽤ dst_neigh_output 继续传递 skb。

//file: include/net/dst.h
static inline int
dst_neigh_output(struct dst_entry *dst, struct neighbour *n, struct sk_buff *skb)
{......return n->output(n, skb);
}

调⽤ output，实际指向的是 neigh_resolve_output。
neigh_resolve_output 内部有可能会发出 arp ⽹络请求。

//file: net/core/neighbour.c
int neigh_resolve_output(){//注意：这⾥可能会触发 arp 请求if (!neigh_event_send(neigh, skb)) {//neigh->ha 是 MAC 地址dev_hard_header(skb, dev, ntohs(skb->protocol), neigh->ha, NULL, skb->len);//发送dev_queue_xmit(skb);}
}

当获取到硬件 MAC 地址以后，就可以封装 skb 的 MAC 头了。最后调⽤ dev_queue_xmit 将 skb 传递给 Linux ⽹络设备⼦系统。

网络设备子系统

邻居⼦系统通过 dev_queue_xmit 进⼊到⽹络设备⼦系统中来。

//file: net/core/dev.c
int dev_queue_xmit(struct sk_buff *skb)
{//选择发送队列txq = netdev_pick_tx(dev, skb);//获取与此队列关联的排队规则q = rcu_dereference_bh(txq->qdisc);//如果有队列，则调⽤__dev_xmit_skb 继续处理数据if (q->enqueue) {rc = __dev_xmit_skb(skb, q, dev, txq);goto out;}//没有队列的是回环设备和隧道设备......
}

⽹卡是有多个发送队列的（尤其是现在的⽹卡）。上⾯对 netdev_pick_tx 函数的调⽤就是选择⼀个队列进⾏发送。
netdev_pick_tx 发送队列的选择受 XPS 等配置的影响，⽽且还有缓存，也是⼀套⼩复杂的逻辑。
这⾥我们只关注两个逻辑，⾸先会获取⽤户的 XPS 配置，否则就⾃动计算了。
代码⻅ netdev_pick_tx => __netdev_pick_tx。

//file: net/core/flow_dissector.c
u16 __netdev_pick_tx(struct net_device *dev, struct sk_buff *skb)
{//获取 XPS 配置int new_index = get_xps_queue(dev, skb);//⾃动计算队列if (new_index < 0)new_index = skb_tx_hash(dev, skb);}

然后获取与此队列关联的 qdisc。在 linux 上通过 tc 命令可以看到 qdisc 类型，例如对于我的某台多队列⽹卡机器上是 mq disc。

#tc qdisc
qdisc mq 0: dev eth0 root

⼤部分的设备都有队列（回环设备和隧道设备除外），所以现在我们进⼊到 __dev_xmit_skb。

//file: net/core/dev.c
static inline int __dev_xmit_skb(struct sk_buff *skb, structQdisc *q, struct net_device *dev, struct netdev_queue *txq)
{//1.如果可以绕开排队系统if ((q->flags & TCQ_F_CAN_BYPASS) && !qdisc_qlen(q) && qdisc_run_begin(q)) {......}//2.正常排队else {//⼊队q->enqueue(skb, q)//开始发送__qdisc_run(q);}
}

上述代码中分两种情况，1 是可以 bypass（绕过）排队系统的，另外⼀种是正常排队。我们只看第⼆种情况。
先调⽤ q->enqueue 把 skb 添加到队列⾥。然后调⽤ __qdisc_run 开始发送。

//file: net/sched/sch_generic.c
void __qdisc_run(struct Qdisc *q)
{int quota = weight_p;//循环从队列取出⼀个 skb 并发送while (qdisc_restart(q)) {// 如果发⽣下⾯情况之⼀，则延后处理：// 1. quota ⽤尽// 2. 其他进程需要 CPUif (--quota <= 0 || need_resched()) {//将触发⼀次 NET_TX_SOFTIRQ 类型 softirq__netif_schedule(q);break;}}
}

在上述代码中，我们看到 while 循环不断地从队列中取出 skb 并进⾏发送。注意，这个时候其实都占⽤的是⽤户进程的系统态时间(sy)。 只有当 quota ⽤尽或者其它进程需要 CPU 的时候才触发软中断进⾏发送。
所以这就是为什么⼀般服务器上查看 /proc/softirqs，⼀般 NET_RX 都要⽐ NET_TX ⼤的多的第⼆个原因。
对于读来说，都是要经过 NET_RX 软中断，⽽对于发送来说，只有系统态配额⽤尽才让软中断上。
再看 qdisc_restart 上，继续看发送过程。

static inline int qdisc_restart(struct Qdisc *q)
{//从 qdisc 中取出要发送的 skbskb = dequeue_skb(q);...return sch_direct_xmit(skb, q, dev, txq, root_lock);
}

qdisc_restart 从队列中取出⼀个 skb，并调⽤ sch_direct_xmit 继续发送。

//file: net/sched/sch_generic.c
int sch_direct_xmit(struct sk_buff *skb, struct Qdisc *q,struct net_device *dev, struct netdev_queue *txq,spinlock_t *root_lock)
{//调⽤驱动程序来发送数据ret = dev_hard_start_xmit(skb, dev, txq);
}

软中断调度

如果系统态 CPU 发送⽹络包不够⽤的时候，会调⽤ __netif_schedule 触发⼀个软中断。该函数会进⼊到 __netif_reschedule，由它来实际发出 NET_TX_SOFTIRQ 类型软中断。
软中断是由内核线程来运⾏的，该线程会进⼊到 net_tx_action 函数，在该函数中能获取到发送队列，并也最终调⽤到驱动程序⾥的⼊⼝函数 dev_hard_start_xmit。

//file: net/core/dev.c
static inline void __netif_reschedule(struct Qdisc *q)
{sd = &__get_cpu_var(softnet_data);q->next_sched = NULL;*sd->output_queue_tailp = q;sd->output_queue_tailp = &q->next_sched;......raise_softirq_irqoff(NET_TX_SOFTIRQ);
}

在该函数⾥在软中断能访问到的 softnet_data ⾥设置了要发送的数据队列，添加到了 output_queue ⾥了。紧接着触发了 NET_TX_SOFTIRQ 类型的软中断。

NET_TX_SOFTIRQ softirq 注册的回调函数 net_tx_action，⽤户态进程触发完软中断之后，会有⼀个软中断内核线程会执⾏到 net_tx_action。
注意：这以后发送数据消耗的 CPU 就都显示在 si 这⾥了，不会消耗⽤户进程的系统时间了。

//file: net/core/dev.c
static void net_tx_action(struct softirq_action *h)
{//通过 softnet_data 获取发送队列struct softnet_data *sd = &__get_cpu_var(softnet_data);// 如果 output queue 上有 qdiscif (sd->output_queue) {// 将 head 指向第⼀个 qdischead = sd->output_queue;//遍历 qdsics 列表while (head) {struct Qdisc *q = head;head = head->next_sched;//发送数据qdisc_run(q);}}
}

软中断这⾥会获取 softnet_data。前⾯我们看到进程内核态在调⽤ __netif_reschedule 的时候把发送队列写到 softnet_data 的 output_queue ⾥了。 软中断循环遍历 sd->output_queue 发送数据帧。
来看 qdisc_run，它和进程⽤户态⼀样，也会调⽤到 __qdisc_run。

//file: include/net/pkt_sched.h
static inline void qdisc_run(struct Qdisc *q)
{if (qdisc_run_begin(q))__qdisc_run(q);
}

然后⼀样就是进⼊ qdisc_restart => sch_direct_xmit，直到驱动程序函数 dev_hard_start_xmit。

igb网卡驱动发送

我们前⾯看到，⽆论是对于⽤户进程的内核态，还是对于软中断上下⽂，都会调⽤到⽹络设备⼦系统中的 dev_hard_start_xmit 函数。在这个函数中，会调⽤到驱动⾥的发送函数 igb_xmit_frame。
在驱动函数⾥，将 skb 会挂到 RingBuffer上，驱动调⽤完毕后，数据包将真正从⽹卡发送出去。

//file: net/core/dev.c
int dev_hard_start_xmit(struct sk_buff *skb, struct net_device *dev, struct netdev_queue *txq)
{//获取设备的回调函数集合 opsconst struct net_device_ops *ops = dev->netdev_ops;//获取设备⽀持的功能列表features = netif_skb_features(skb);//调⽤驱动的 ops ⾥⾯的发送回调函数 ndo_start_xmit 将数据包传给⽹卡设备skb_len = skb->len;rc = ops->ndo_start_xmit(skb, dev);
}

其中 ndo_start_xmit 是⽹卡驱动要实现的⼀个函数，是在 net_device_ops 中定义的。

//file: include/linux/netdevice.h
struct net_device_ops {netdev_tx_t (*ndo_start_xmit) (struct sk_buff *skb, struct net_device *dev);
}

在 igb ⽹卡驱动源码中，我们找到了。

//file: drivers/net/ethernet/intel/igb/igb_main.c
static const struct net_device_ops igb_netdev_ops = {.ndo_open = igb_open,.ndo_stop = igb_close,.ndo_start_xmit = igb_xmit_frame,...
};

也就是说，对于⽹络设备层定义的 ndo_start_xmit， igb 的实现函数是 igb_xmit_frame。
这个函数是在⽹卡驱动初始化的时候被赋值的。，所以在上⾯⽹络设备层调⽤ ops->ndo_start_xmit 的时候，会实际上进⼊ igb_xmit_frame 这个函数中。我们进⼊这个函数来看看驱动程序是如何⼯作的。

//file: drivers/net/ethernet/intel/igb/igb_main.c
static netdev_tx_t igb_xmit_frame(struct sk_buff *skb,struct net_device *netdev)
{......return igb_xmit_frame_ring(skb, igb_tx_queue_mapping(adapter, skb));
}
netdev_tx_t igb_xmit_frame_ring(struct sk_buff *skb, struct igb_ring *tx_ring)
{    //获取TX Queue 中下⼀个可⽤缓冲区信息first = &tx_ring->tx_buffer_info[tx_ring->next_to_use];first->skb = skb;first->bytecount = skb->len;first->gso_segs = 1;//igb_tx_map 函数准备给设备发送的数据。igb_tx_map(tx_ring, first, hdr_len);
}

在这⾥从⽹卡的发送队列的 RingBuffer 中取下来⼀个元素，并将 skb 挂到元素上。

igb_tx_map 函数处理将 skb 数据映射到⽹卡可访问的内存 DMA 区域。

//file: drivers/net/ethernet/intel/igb/igb_main.c
static void igb_tx_map(struct igb_ring *tx_ring,struct igb_tx_buffer *first, const u8 hdr_len)
{//获取下⼀个可⽤描述符指针tx_desc = IGB_TX_DESC(tx_ring, i);//为 skb->data 构造内存映射，以允许设备通过 DMA 从 RAM 中读取数据dma = dma_map_single(tx_ring->dev, skb->data, size, DMA_TO_DEVICE);//遍历该数据包的所有分⽚,为 skb 的每个分⽚⽣成有效映射for (frag = &skb_shinfo(skb)->frags[0];; frag++) {tx_desc->read.buffer_addr = cpu_to_le64(dma);tx_desc->read.cmd_type_len = ...;tx_desc->read.olinfo_status = 0;}//设置最后⼀个descriptorcmd_type |= size | IGB_TXD_DCMD;tx_desc->read.cmd_type_len = cpu_to_le32(cmd_type);/* Force memory writes to complete before letting h/w know there* are new descriptors to fetch*/wmb();
}

当所有需要的描述符都已建好，且 skb 的所有数据都映射到 DMA 地址后，驱动就会进⼊到它的最后⼀步，触发真实的发送。

发送完成硬件中断

当数据发送完成以后，其实⼯作并没有结束。因为内存还没有清理。当发送完成的时候，⽹卡设备会触发⼀个硬中断来释放内存。
在发送硬中断⾥，会执⾏ RingBuffer 内存的清理⼯作，如图：

再回头看⼀下硬中断触发软中断的源码。

//file: drivers/net/ethernet/intel/igb/igb_main.c
static inline void ____napi_schedule(...){list_add_tail(&napi->poll_list, &sd->poll_list);__raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

这⾥有个很有意思的细节，⽆论硬中断是因为是有数据要接收，还是说发送完成通知，从硬中断触发的软中断都是 NET_RX_SOFTIRQ。 这个我们在第⼀节说过了，这是软中断统计中 RX 要⾼于 TX 的⼀个原因。
好我们接着进⼊软中断的回调函数 igb_poll。在这个函数⾥，我们注意到有⼀⾏ igb_clean_tx_irq，参⻅源码：

//file: drivers/net/ethernet/intel/igb/igb_main.c
static int igb_poll(struct napi_struct *napi, int budget)
{//performs the transmit completion operationsif (q_vector->tx.ring)clean_complete = igb_clean_tx_irq(q_vector);...
}

我们来看看当传输完成的时候，igb_clean_tx_irq 都⼲啥了。

//file: drivers/net/ethernet/intel/igb/igb_main.c
static bool igb_clean_tx_irq(struct igb_q_vector *q_vector)
{//free the skbdev_kfree_skb_any(tx_buffer->skb);//clear tx_buffer datatx_buffer->skb = NULL;dma_unmap_len_set(tx_buffer, len, 0);// clear last DMA location and unmap remaining buffers */while (tx_desc != eop_desc) {}
}

⽆⾮就是清理了 skb，解除了 DMA 映射等等。 到了这⼀步，传输才算是基本完成了。
为啥我说是基本完成，⽽不是全部完成了呢？因为传输层需要保证可靠性，所以 skb 其实还没有删除。它得等收到对⽅的 ACK 之后才会真正删除，那个时候才算是彻底的发送完毕。

五、问题

1、我们在监控内核发送数据消耗的 CPU 时，是应该看 sy 还是 si ？
在⽹络包的发送过程中，⽤户进程（在内核态）完成了绝⼤部分的⼯作，甚⾄连调⽤驱动的事情都⼲了。 只有当内核态进程被切⾛前才会发起软中断。 发送过程中，绝⼤部分（90%）以上的开销都是在⽤户进程内核态消耗掉的。
只有⼀少部分情况下才会触发软中断（NET_TX 类型），由软中断 ksoftirqd 内核进程来发送。
所以，在监控⽹络 IO 对服务器造成的 CPU 开销的时候，不能仅仅只看 si，⽽是应该把 si、sy 都考虑进来。
2. 在服务器上查看 /proc/softirqs，为什么 NET_RX 要⽐ NET_TX ⼤的多的多？
之前我认为 NET_RX 是读取，NET_TX 是传输。对于⼀个既收取⽤户请求，⼜给⽤户返回的 Server 来说。 这两块的数字应该差不多才对，⾄少不会有数量级的差异。但事实上，⻜哥⼿头的⼀台服务器是这样的：

经过源码分析，发现这个问题的原因有两个。
（1）当数据发送完成以后，通过硬中断的⽅式来通知驱动发送完毕。但是硬中断⽆论是有数据接收，还是对于发送完毕，触发的软中断都是 NET_RX_SOFTIRQ，⽽并不是 NET_TX_SOFTIRQ。
（2）对于读来说，都是要经过 NET_RX 软中断的，都⾛ ksoftirqd 内核进程。⽽对于发送来说，绝⼤部分⼯作都是在⽤户进程内核态处理了，只有系统态配额⽤尽才会发出 NET_TX，让软中断上。
3.发送⽹络数据的时候都涉及到哪些内存拷⻉操作？
这⾥的内存拷⻉，我们只特指待发送数据的内存拷⻉。
第⼀次拷⻉操作是内核申请完 skb 之后，这时候会将⽤户传递进来的 buffer ⾥的数据内容都拷⻉到 skb 中。如果要发送的数据量⽐较⼤的话，这个拷⻉操作开销还是不⼩的。
第⼆次拷⻉操作是从传输层进⼊⽹络层的时候，每⼀个 skb 都会被克隆⼀个新的副本出来。
⽹络层以及下⾯的驱动、软中断等组件在发送完成的时候会将这个副本删除。传输层保存着原始的 skb，在当⽹络对⽅没有 ack 的时候，还可以重新发送，以实现 TCP 中要求的可靠传输。
第三次拷⻉不是必须的，只有当 IP 层发现 skb ⼤于 MTU 时才需要进⾏。会再申请额外的 skb，并将原来的 skb 拷⻉为多个⼩的 skb。
在⽹络性能优化中经常听到的零拷⻉，TCP 为了保证可靠性，第⼆次的拷⻉根本就没法省。如果包再⼤于 MTU 的话，分⽚时的拷⻉同样也避免不了。