深入理解Linux网络笔记（四）：内核是如何发送网络包的

网络笔记（四）：内核是如何发送网络包的"/>

深入理解Linux网络笔记（四）：内核是如何发送网络包的

本文为《深入理解Linux网络》学习笔记，使用的Linux源码版本是3.10，网卡驱动默认采用的都是Intel的igb网卡驱动

Linux源码在线阅读：.10/source

3、内核是如何发送网络包的

1）、网络包发送过程总览

int main() {fd = socket(AF_INET, SOCK_STREAM, 0);bind(fd, ...);listen(fd, ...);cfd = accept(fd, ...);// 接收用户请求read(cfd, ...);// 用户请求处理dosomething();// 给用户返回结果send(cfd, buf, sizeof(buf), 0);
}

上述代码中，调用send之后内核是怎样把数据包发送出去的？

如上图所示，可以看到用户数据被拷贝到内核态，然后经过协议栈处理后进入RingBuffer。随后网卡驱动真正将数据发送了出去。当发送完成的时候，是通过硬中断来通知CPU，然后清理RingBuffer

2）、网卡启动准备

现在的服务器上的网卡一般都是支持多队列的。每一个队列都是由一个RingBuffer表示的，开启了多队列以后的网卡就会对应有多个RingBuffer，如下图所示：

网卡在启动时最重要的任务之一就是分配和初始化RingBuffer，接下来看看网卡启动时分配传输队列RingBuffer的实际过程

在网卡启动的时候，会调用__igb_open函数，RingBuffer就是在这里分配的

// 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

// 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++) {err = igb_setup_tx_resources(adapter->tx_ring[i]);...}...
}

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

// 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的内部也不仅仅是一个环形队列数组：

• igb_tx_buffer数组：这个数组是内核使用的，通过vzalloc申请
• e1000_adv_tx_desc数组：这个数组是网卡硬件使用的，通过dma_alloc_coherent分配

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

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

3）、数据从用户进程到网卡的详细过程

1）send系统调用实现

send系统调用的源码位于net/socket.c中。在这个系统调用里，内部其实真正使用的是sendto系统调用。主要干了两件事情：

1. 在内核中把真正的socket找出来，在这个对象里记录着各种协议栈的函数地址
2. 构造一个struct msghdr对象，把用户传入的数据，比如buffer地址、数据长度什么的，都装进去

剩下的事情就交给下一层，协议栈里的函数inet_sendmsg了，其中inet_sendmsg函数的地址是通过socket内核对象里的ops成员找到的。大致流程如下图所示：

// 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(sendto, int, fd, void __user *, buff, size_t, len,unsigned int, flags, struct sockaddr __user *, addr,int, addr_len)
{struct socket *sock;...struct msghdr msg;struct iovec iov;...// 1.根据fd找到socketsock = sockfd_lookup_light(fd, &err, &fput_needed);// 2.构造msghdriov.iov_base = buff;iov.iov_len = len;msg.msg_name = NULL;msg.msg_iov = &iov;msg.msg_iovlen = 1;...msg.msg_flags = flags;// 3.发送数据err = 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中，函数调用将会由系统调用进入协议栈，源码如下：

// net/socket.c
static inline int __sock_sendmsg_nosec(struct kiocb *iocb, struct socket *sock,struct msghdr *msg, size_t size)
{...return sock->ops->sendmsg(iocb, sock, msg, size);
}

socket内核对象结构图：

通过上图可以看到，__sock_sendmsg_nosec中调用的是sock->ops->sendmsg，实际执行的是inet_sendmsg。这个函数是AF_INET协议族提供的通用发送函数

2）传输层处理

传输层拷贝

在进入协议栈inet_sendmsg以后，内核接着会找到socket上的具体协议发送函数。对于TCP协议来说，那就是tcp_sendmsg（同样也是通过socket内核对象找到的）

在这个函数中，内核会申请一个内核态的skb内存，将用户待发送的数据拷贝进去。注意，这个时候不一定会真正开始发送，如果没有达到发送条件，很可能这次调用直接就返回了，大概过程如下图所示：

inet_sendmsg函数源码如下：

// net/ipv4/af_inet.c
int inet_sendmsg(struct kiocb *iocb, struct socket *sock, struct msghdr *msg,size_t size)
{...return sk->sk_prot->sendmsg(iocb, sk, msg, size);
}

在这个函数中会调用到具体协议的发送函数。参考上面的socket内核对象结构图，可以看到对于TCP下的socket来说，sk->sk_prot->sendmsg指向的是tcp_sendmsg

tcp_sendmsg这个函数比较长，分成多块来看。先看以下这一段：

// net/ipv4/tcp.c
int tcp_sendmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,size_t size)
{...while (--iovlen >= 0) {...while (seglen > 0) {...// 获取发送队列skb = tcp_write_queue_tail(sk);// 申请skb并拷贝...}}...
}

// include/net/tcp.h
static inline struct sk_buff *tcp_write_queue_tail(const struct sock *sk)
{return skb_peek_tail(&sk->sk_write_queue);
}

tcp_write_queue_tail函数在获取socket发送队列中的最后一个skb。skb是struct sk_buff对象的简称，用户的发送队列就是该对象组成的一个链表，如下图所示：

接着看tcp_sendmsg的其他部分

// net/ipv4/tcp.c
int tcp_sendmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,size_t size)
{...// 获取用户传递过来的数据和标志flags = msg->msg_flags; // 各种标志...iovlen = msg->msg_iovlen; // 数据块数为1iov = msg->msg_iov; // 用户数据地址...// 遍历用户层的数据块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是用户空间的数据地址err = skb_add_data_nocache(sk, skb, from, copy);...}...}}...
}

这个函数比较长，不过其实逻辑并不复杂。其中msg->msg_iov存储的是用户态内存要发送的数据的buffer。接下来在内核态申请内核内存，比如skb，并把用户内存里的数据拷贝到内核态内存中，如下图所示。这就会涉及一次或者几次内存拷贝的开销

至于内核什么时候真正把skb发送出去，在tcp_sendmsg中会进行一些判断

// net/ipv4/tcp.c
int tcp_sendmsg(struct kiocb *iocb, struct sock *sk, struct msghdr *msg,size_t size)
{...while (--iovlen >= 0) {...while (seglen > 0) {// 申请内核内存并进行拷贝...// 发送判断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传到更低的网络层进行处理。传输层发送流程总图如下图所示：

tcp_write_xmit源码如下：

// 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);...if (unlikely(!tcp_snd_wnd_test(tp, skb, mss_now)))break;...if (skb->len > limit &&unlikely(tso_fragment(sk, skb, limit, mss_now, gfp)))break;...// 真正开启发送if (unlikely(tcp_transmit_skb(sk, skb, 1, gfp)))break;...}...
}

在网络协议里学的滑动窗口、拥塞控制就是在这个函数中完成的。这里只看发送主流程，那就走到了tcp_transmit_skb

// 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)) {...if (unlikely(skb_cloned(skb)))skb = pskb_copy(skb, gfp_mask);elseskb = skb_clone(skb, gfp_mask);...}...// 2.封装TCP头th = tcp_hdr(skb);th->source		= inet->inet_sport;th->dest		= inet->inet_dport;...if (unlikely(tcb->tcp_flags & TCPHDR_SYN)) {th->window	= htons(min(tp->rcv_wnd, 65535U));} else {th->window	= htons(tcp_select_window(sk));}...// 3.调用网络层发送接口err = icsk->icsk_af_ops->queue_xmit(skb, &inet->cork.fl);...
}

第一件事是先克隆一个新的skb，为什么要复制一个skb出来？

这是因为skb后续在调用网络层，最后到达网卡发送完成的时候，这个skb会被释放掉。而TCP协议是支持丢失重传的，在收到对方的ACK之前，这个skb不能被删除。所以内核的做法就是每次调用网卡发送的时候，实际上传递出去的是skb的一个拷贝。等收到ACK再真正删除

第二件事是修改skb中的TCP头，根据实际情况把TCP头设置好。skb内部其实包含了网络协议中所有的头。在设置TCP头的时候，只是把指针指向skb的合适位置。后面设置IP头的时候，再把指针挪一挪就行，如下图所示。避免频繁的内存申请和拷贝，效率很高

tcp_transmit_skb是发送数据位于传输层的最后一步，接下来就可以进入网络层进行下一层的操作了。调用了网络层提供的发送接口icsk->icsk_af_ops->queue_xmit()

在下面这个源码选中，可以看到queue_xmit其实指向的是ip_queue_xmit函数

// 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,...
}

自此，传输层的工作也就都完成了。数据离开了传输层，接下来会进入内核在网络层的实现

3）网络层发送处理

Linux内核网络层的发送的实现位于net/ipv4/ip_output.c这个文件。传输层调用到的ip_queue_xmit也在这里

在网络层主要处理路由项查找、IP头设置、netfilter过滤、skb切分（大于MTU的话）等几项工作，处理完这些工作后会交给更下一层的邻居子系统来处理。网络层发送处理过程如下图：

网络层入口函数ip_queue_xmit源码如下：

// 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(sock_net(sk), fl4, sk,daddr, inet->inet_saddr,inet->inet_dport,inet->inet_sport,sk->sk_protocol,RT_CONN_FLAGS(sk),sk->sk_bound_dev_if);...sk_setup_caps(sk, &rt->dst);}// 为skb设置路由表skb_dst_set_noref(skb, &rt->dst);...// 设置IP头iph = ip_hdr(skb);...iph->ttl      = ip_select_ttl(inet, &rt->dst);iph->protocol = sk->sk_protocol;...// 发送res = ip_local_out(skb);...
}

ip_queue_xmit已经到了网络层，在这个函数里我们看到了网络层的功能路由项查找，如果找到了则设置到skb上（没有路由的话就直接报错返回了）

在Liunx上通过route命令可以看到本机的路由配置，如下：

$ route -n
Kernel IP routing table
Destination     Gateway         Genmask         Flags Metric Ref    Use Iface
10.0.0.0        10.162.x.x      255.0.0.0       UG    0    	 0        0 eth0
10.162.x.0		0.0.0.0			255.255.248.0	U	  0		 0		  0 eth0
169.x.x.x		0.0.0.0			255.255.0.0		U	  1002	 0		  0 eth0	

在路由表中，可以查到某个目的网络应该通过哪个Iface（网卡）、哪个Gateway（网关）发送出去。查找出来以后缓存到socket上，下次再发送数据就不用查了

接着把路由表地址也放到skb里

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

接下来就是定位到skb里的IP头的位置，然后开始按照协议规范设置IP头，如下图所示：

再通过ip_local_out进入下一步的处理

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

在调用ip_local_out => __ip_local_out => nf_hook的过程中会执行netfilter过滤。如果使用iptables配置了一些规则，那么这里将检测是否命中规则。如果设置了非常复杂的netfilter规则，在这里这个函数将会导致进程CPU开销大增

继续看和发送有关的过程dst_output

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

此函数找到这个skb的路由表（dst条目），然后调用路由表的output方法。这又是一个函数指针，指向的ip_output方法

// 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

// 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字节。可以通过控制程序的数据包尺寸小于MTU的方式来优化网络性能。因为分片会带来两个问题：

1. 需要进行额外的切分处理，有额外性能开销
2. 只要一个分片丢失，整个包都要重传

所以避免分片既杜绝了分片开销，也大大降低了重传率

在ip_finish_output2中，发送过滤终于进入下一层：邻居子系统

// 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);if (!IS_ERR(neigh)) {// 继续向下层传递int res = dst_neigh_output(dst, neigh, skb);...}...
}

4）邻居子系统

邻居子系统是位于网络层和数据链路层中间的一个系统，其作用是为网络层提供一个下层的封装，让网络层不必关心下层的地址信息，让下层来决定发送到哪个MAC地址

这个邻居子系统位于net/core/neighbour.c，无论是对于IPv4还是IPv6，都需要使用该模块，如下图所示：

在邻居子系统里主要查找或者创建邻居项，在创建邻居项的时候，有可能会发出实际的arp请求。然后封装MAC头，将发送过程再传递到更下层的网络设备子系统。大致流程如下图所示：

在上面的ip_finish_output2源码中调用了__ipv4_neigh_lookup_noref。它在arp缓存中进行查找，其第二个参数传入的是路由下一跳IP信息

// include/net/arp.h
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);struct neighbour *n;u32 hash_val;// 计算哈希值,加速查找hash_val = arp_hashfn(key, dev, nht->hash_rnd[0]) >> (32 - nht->hash_shift);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;}return NULL;
}

如果找不到，则调用__neigh_create创建一个邻居

// net/core/neighbour.c
struct neighbour *__neigh_create(struct neigh_table *tbl, const void *pkey,struct net_device *dev, bool want_ref)
{...// 申请邻居表项struct neighbour *n1, *rc, *n = neigh_alloc(tbl, dev);...// 构造赋值memcpy(n->primary_key, pkey, key_len);n->dev = dev;...if (n->parms->neigh_setup &&(error = n->parms->neigh_setup(n)) < 0) {rc = ERR_PTR(error);goto out_neigh_release;}...// 最后添加到邻居哈希表中rcu_assign_pointer(nht->hash_buckets[hash_val], n);...
}

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

// 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。在这个函数内部有可能发出arp网路请求

// net/core/neighbour.c
int neigh_resolve_output(struct neighbour *neigh, struct sk_buff *skb)
{...// 这里可能会触发arp请求if (!neigh_event_send(neigh, skb)) {...do {__skb_pull(skb, skb_network_offset(skb));seq = read_seqbegin(&neigh->ha_lock);// neigh->ha是MAC地址err = dev_hard_header(skb, dev, ntohs(skb->protocol),neigh->ha, NULL, skb->len);} while (read_seqretry(&neigh->ha_lock, seq));if (err >= 0)// 发送rc = dev_queue_xmit(skb);...}...
}

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

5）网络设备子系统

邻居子系统通过dev_queue_xmit进入网络设备子系统。网络设备子系统的工作流程如下图所示：

dev_queue_xmit源码如下：

// 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函数

// 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 qdisc

$ tc qdisc
qdisc mq 0: dev eth0 root

大部分的设备都有队列（回环设备和隧道设备除外），所以现在进入__dev_xmit_skb

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

上述代码中分两种情况，一种是可以bypass（绕过）排队系统，另外一种是正常排队。只看第二种情况

先调用q->enqueue把skb添加到队列里，然后调用__qdisc_run开始发送

// 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;}}qdisc_run_end(q);
}

在上述代码中可以看到，while循环不断地从队列中取出skb并进行发送。注意，这个时候其实都占用的是用户进程的系统态时间（sy）。只有当quota用尽或者其他进程需要CPU的时候才触发软中断进行发送

所以这就是为什么在服务器上查看/proc/softirqs，一般NET_RX都要比NET_TX大得多的原因之一。对于接收来说，都要经过NET_RX软中断，而对于发送来说，只有系统态配额用尽才让软中断上

qdisc_restart源码如下：

// net/sched/sch_generic.c
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继续发送

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

6）软中断调度

__qdisc_run函数中看到，如果发送网络包的时候系统态CPU用尽了，会调用__netif_schedule触发一个软中断。该函数会进入__netif_reschedule，由它来实际发出NET_TX_SOFTIRQ类型软中断

软中断是由内核进程来运行的，该进程会进入net_tx_action函数，在该函数中能获取发送队列，并也最终调用到驱动程序里的入口函数dev_hard_start_xmit，如下图所示：

// net/core/dev.c
static inline void __netif_reschedule(struct Qdisc *q)
{struct softnet_data *sd;unsigned long flags;local_irq_save(flags);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);local_irq_restore(flags);
}

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

之前的文章已经介绍过软中断代码了，这里直接从NET_TX_SOFTIRQ softirq注册的回调函数net_tx_action讲起。用户态进程触发完软中断之后，会有一个软中断内核线程执行到net_tx_action

这以后发送数据消耗的CPU就都显示在si这里，不会消耗用户进程的系统时间

// 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;...// 遍历qdisc列表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

// 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

7）igb网卡驱动发送

通过前面的介绍可知，无论对于用户进程的内核态，还是对于软中断上下文，都会调用网络设备子系统中的dev_hard_start_xmit函数。在这个函数中，会调用驱动里的发送函数igb_xmit_frame

在驱动函数里，会将skb挂到RingBuffer上，驱动调用完毕，数据包将真正从网卡发送出去。网卡驱动工作流程如下图所示：

// 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;...if (likely(!skb->next)) {...// 获取设备支持的功能列表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中定义的

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

在igb网卡驱动源码中找到了net_device_ops函数

// 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这个函数

igb_xmit_frame函数代码如下：

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

// drivers/net/ethernet/intel/igb/igb_main.c
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区域

// 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 = cpu_to_le32(cmd_type ^ size);...tx_desc->read.olinfo_status = 0;...}// 设置最后一个descriptorcmd_type |= size | IGB_TXD_DCMD;tx_desc->read.cmd_type_len = cpu_to_le32(cmd_type);...
}

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

4）、RingBuffer内存回收

当数据发送完以后，其实工作并没有结束，因为内存还没有清理。当发送完成的时候，网卡设备会触发一个硬中断来释放内存。在发送硬中断的过程里，会执行RingBuffer内存的清理工作，如下图所示：

硬中断触发软中断的源码如下：

// net/core/dev.c
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);
}

无论是硬中断是因为有数据要接收，还是发送完成通知，从硬中断触发的软中断都是NET_RX_SOFTIRQ。它也是软中断统计中RX要高于TX的一个原因

接着进入软中断的回调函数igb_poll。在这个函数里，有一行igb_clean_tx_irq，参见以下源码：

// drivers/net/ethernet/intel/igb/igb_main.c
static int igb_poll(struct napi_struct *napi, int budget)
{...if (q_vector->tx.ring)clean_complete = igb_clean_tx_irq(q_vector);...
}

当传输完成的时候，igb_clean_tx_irq都干了什么

// drivers/net/ethernet/intel/igb/igb_main.c
static bool igb_clean_tx_irq(struct igb_q_vector *q_vector)
{...do {...// 释放skbdev_kfree_skb_any(tx_buffer->skb);...// 清除tx_buffer数据tx_buffer->skb = NULL;dma_unmap_len_set(tx_buffer, len, 0);// 清除最后的DMA位置,解除映射while (tx_desc != eop_desc) {...}...} while (likely(budget));...
}

无非就是清理了skb，解除了DMA映射等等。到了这一步，传输才算是基本完成了

为什么说是基本完成，而不是全部完成了呢？因为传输层需要保证可靠性，所以skb其实还没有删除。它得等收到对方的ACK之后才会真正删除，那个时候才算彻底发送完毕

5）、总结

整个发送过程总结如下图：

1）我们在监控内核发送数据消耗的CPU时，应该看sy还是si？

在网络包的发送过程中，用户进程（在内核态）完成了绝大部分的工作，甚至连调用驱动的工作都干了。只当内核态进程被切走前才会发起软中断。发送过程中，绝大部分（90%）以上的开销都是用户进程内核态消耗掉的

只有一少部分情况下才会触发软中断（NET_TX类型），由软中断ksoftirqd内核线程来发送

所以，在监控网络IO对服务器造成的CPU开销的时候，不能仅看si，而是应该把si、sy都考虑进来

2）在服务器上查看/proc/softirqs，为什么NET_RX要比NET_TX大得多的多？

之前认为NET_RX是接收，NET_TX是传输。对于一个既收取用户请求，用给用户返回的服务器来说，这两块的数字应该差不多才对，至少不会有数量级的差异。但事实上，一台服务器情况如下图：

造成这个问题的原因有两个：

1. 当数据发送完以后，通过硬中断的方式来通知驱动发送完毕。但是硬中断无论是有数据接收，还是发送完毕，触发的软中断都是NET_RX_SOFTIRQ，并不是NET_TX_SOFTIRQ
2. 对于读来说，都是要经过NET_RX软中断的，都走ksoftirqd内核线程。而对于发送来说，绝大部分工作都是在用户进程内核态处理了，只有系统态配额用尽才会发出NET_TX，让软中断上

3）发送网络数据的时候都涉及哪些内存拷贝操作？

这里的内存拷贝，只特指待发送数据的内存拷贝

第一次拷贝操作是在内核申请完skb以后，这时候会将用户传递进来的buffer里的数据内容都拷贝到skb。如果要发送的数据量比较大，这个拷贝操作开销还是不小的

第二次拷贝操作是从传输层进入网络层的时候，每一个skb都会被克隆出来一个新的副本。目的是保存原始的skb，当网络对方没有发回ACK的时候，还可以重新发送，以实现TCP中要求的可靠传输。不过这次这是浅拷贝，只拷贝skb描述符本身，所指向的数据还是复用的

第三次拷贝不是必须的，只有当IP层发现skb大于MTU时才需要进行。此时会再申请额外的skb，并将原来的skb拷贝成多个小的skb

零拷贝主要是节省了第一次拷贝的开销。TCP为了保证可靠性，第二次的拷贝根本没法省。如果包大于MTU，分片时的拷贝同样避免不了

4）零拷贝到底是怎么回事？

如果把本机的一个文件通过网络发送出去，做法之一就是先用read系统调用把文件读取到内存，然后再调用send把文件发送出去

假设数据之前从来没有读取过，那么read硬盘上的数据需要经过两次拷贝才能到用户进程的内存。第一次是从硬盘DMA到Page Cache。第二次是从Page Cache拷贝到用户内存。send系统调用在前面讲过了。那么read+send系统调用发送一个文件出去数据需要经过的拷贝过程如下图所示：

如果要发送的数据量比较大，那需要花费不少的时间在大量的数据拷贝上。前面提到的sendfile就是内核提供的一个可用来减少发送文件时拷贝开销的一个技术方案。在sendfile系统调用里，数据不需要拷贝到用户空间，在内核态就能完成发送处理，如下图所示，这就显著减少了需要拷贝的次数

推荐阅读：

IP数据分片之MTU和TCP的MSS

ARP 协议详解(网络层)

LINUX XPS分析

linux tc流量控制（一）：classless qdisc

linux tc流量控制（二）：classful qdisc

零拷贝技术：mmap和sendfile