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

news2024/11/16 13:46:34

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

Linux源码在线阅读：https://elixir.bootlin.com/linux/v3.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)
{
	...
	// 有几个队列就构造几个RingBuffer
	for (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找到socket
	sock = sockfd_lookup_light(fd, &err, &fput_needed);
	// 2.构造msghdr
	iov.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; // 数据块数为1
	iov = msg->msg_iov; // 用户数据地址
	...
	// 遍历用户层的数据块
	while (--iovlen >= 0) {
		...
		// 待发送数据块的地址
		unsigned char __user *from = iov->iov_base;
		...
		while (seglen > 0) {
			...
			// 需要申请新的skb
			if (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)
{
	...
	// 循环获取待发送skb
	while ((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);
		else
			skb = 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_output
	return 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);
	else
		return 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.其他进程需要CPU
		if (--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中取出要发送的skb
	skb = 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上有qdisc
	if (sd->output_queue) {
		...
		// 将head指向第一个qdisc
		head = 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)
{
	// 获取设备的回调函数集合ops
	const 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 {
	...
	netdev_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;
		...
	}

	// 设置最后一个descriptor
	cmd_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 {
		...
		// 释放skb
		dev_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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1139376.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

聚观早报 |2024款飞凡R7官宣；小米14新配色材质

聚观早报 |2024款飞凡R7官宣；小米14新配色材质

【聚观365】10月27日消息 2024款飞凡R7官宣 小米14新配色材质 金山办公2023第三季度业绩 IBM2023第三季度业绩 新东方2024财年第一季度业绩 2024款飞凡R7官宣 飞凡汽车官宣&#xff0c;2024款飞凡R7将于11月上市&#xff0c;新车将搭载飞凡巴赫座舱&#xff0c;同时超过1…
阅读更多...
LabVIEW应用开发——基本函数(一)

LabVIEW应用开发——基本函数(一)

前面我们介绍了一些控件的介绍和属性的配置&#xff0c;想要完成一个软件只会拖控件肯定是不行的&#xff0c;没办法实现既有的功能。比如我们要实现从串口中读到数据&#xff0c;根据一定的协议解析&#xff0c;然后转换成各个参数的值的显示&#xff0c;包括时间、电压、电流…
阅读更多...
Simulink中如何使用数组以及创建数组、对数组索引、赋值

Simulink中如何使用数组以及创建数组、对数组索引、赋值

一、总体概述 1、数组的创建&#xff1a;Mux/constant 2、数组的切片&#xff1a;Demux 3、数组的索引&#xff1a;Selector 4、数组的赋值&#xff1a;Assignment 5、数组的运算&#xff1a;数组可只直接使用运算模块、点乘、矩阵运算 二、详细步骤 1、数组的创建 &am…
阅读更多...
【Docker】Linux网络命名空间

【Docker】Linux网络命名空间

命名空间 Namespace是Linux提供的一种对于系统全局资源的隔离机制&#xff1b;从进程的视角来看&#xff0c;同一个namespace中的进程看到的是该namespace自己独立的一份全局资源&#xff0c;这些资源的变化只在本namespace中可见&#xff0c;对其他namespace没有影响。容器就…
阅读更多...
stm32的ADC采样率如何通过Time定时器进行控制

stm32的ADC采样率如何通过Time定时器进行控制

ADC采样率是个跟重要的概念. 手册上说可以通过Timer定时器进行触发ADC采样. 可我这边悲剧的是, 无论怎么样. ADC都会进行采样. 而且就算是TIM停掉也是一样会进行采样. 这就让我摸不着头脑了… 我想通过定时器动态更改ADC的采样频率. 结果不随我愿… 这到底是什么问题呢? 一…
阅读更多...
STM32 pack STM32F系列开发包下载/stm32 Cube Keil V5 开发pack突然丢失

STM32 pack STM32F系列开发包下载/stm32 Cube Keil V5 开发pack突然丢失

文章目录 1.做STM32开发时，经常发现下载的DEMO代码无法打开，ST自带的更新库软件根本连不上服务器，每次到此都非常恼火。即使进入到官方下载，但每次下载时间超级长都是几小时级别的，如果上面的链接满足不了，那就花几个积分，去下边的链接下载吧。1.各种安装包下载链接下载…
阅读更多...
Q41F-40C手动球阀型号解析

Q41F-40C手动球阀型号解析

Q41F-40C型号字母含义解析 Q41F-40C是德特森阀门常用的手动球阀型号字母分别代表的意思是: Q——代表阀门类型《球阀》 4——代表连接方式《法兰》 1——代表结构形式《浮动式》 F——代表阀座材料《聚四氟乙烯》 -《分隔键》 40——代表公称压力《4.0MPA》 C——代表阀…
阅读更多...
红队专题-Web渗透之资产情报信息收集能力(社工)总结

红队专题-Web渗透之资产情报信息收集能力(社工)总结

信息收集 思路框架知识整理 招募六边形战士队员主动信息收集-直接访问[工具]打点收集内容服务器系统版本、域名域名信息收集工具 dnsenumtheHarvesterLayer子域名收集 DiscoverSubdomain子域名信息搜集工具 wydomain目标域名、DNS收集 subDomainsBrute 端口同服旁站/服务/bann…
阅读更多...
分享！10个使用Angular CLI的实战应用技巧

分享！10个使用Angular CLI的实战应用技巧

如果您正在使用Angular&#xff0c;就知道Angular CLI有多有用。它简化了创建和管理Angular项目的过程&#xff0c;凭借其直观的命令行界面&#xff0c;它可以轻松处理复杂的任务&#xff0c;如生成组件、服务、管道、指令等。 但是您知道如何有效地使用Angular CLI吗&#xf…
阅读更多...
git stash的使用方法

git stash的使用方法

git stash的使用方法 应用场景 当我们在开发一个新功能的时候&#xff0c;或者开发到一半&#xff0c;然后就收到了线上master 出现了bug&#xff0c;当分支开发已经进行了或者进行到一半了&#xff0c;这时怎么办呢&#xff1f; 这时解决方案有两种&#xff1a;一种是先先将当…
阅读更多...
JVisualVM工具的使用

JVisualVM工具的使用

本文来说下JVisualVM工具的基本使用 文章目录 内存泄漏和内存溢出分析工具JVisualVM工具使用本文小结 内存泄漏和内存溢出 简单地说&#xff0c;内存泄漏&#xff08;Memory leak&#xff09;就是在内存中有很多对象没有被回收&#xff0c;一直占着内存&#xff1b;而内存溢出就…
阅读更多...
【Docker】Linux网桥连接多个命名空间

【Docker】Linux网桥连接多个命名空间

veth实现了点对点的虚拟连接&#xff0c;可以通过veth连接两个namespace&#xff0c;如果我们需要将3个或者多个namespace接入同一个二层网络时&#xff0c;就不能只使用veth了。 在物理网络中&#xff0c;如果需要连接多个主机&#xff0c;我们会使用bridge&#xff08;网桥&…
阅读更多...
代码随想录算法训练营第三十四天丨 贪心算法part05

代码随想录算法训练营第三十四天丨 贪心算法part05

435. 无重叠区间 思路 为了让区间尽可能的重叠&#xff0c;我按照左边界排序&#xff0c;从左向右记录重叠区间的个数。最后直接返回记录的重叠区间的个数就行了。 本题其实和昨天做的 ‘用最少数量的箭引爆气球’ 的思路整体是一样的&#xff0c;忘记思路可用看昨天的题目笔…
阅读更多...
JavaScript 中 BOM 基础知识有哪些？

JavaScript 中 BOM 基础知识有哪些？

浏览器对象模型&#xff08;Browser Object Model&#xff0c;简称 BOM&#xff09;是 JavaScript 的组成部分之一&#xff0c;BOM 赋予了 JavaScript 程序与浏览器交互的能力。 window 对象是 BOM 的核心&#xff0c;用来表示当前浏览器窗口&#xff0c;其中提供了一系列用来…
阅读更多...
创新领航 | 竹云参编《基于区块链的数据资产评估实施指南》正式发布！

创新领航 | 竹云参编《基于区块链的数据资产评估实施指南》正式发布！

10月25日&#xff0c;由深圳数宝数据服务股份有限公司和深圳职业技术大学提出&#xff0c;中国科学院深圳先进技术研究院、中国电子技术标准化研究院、中国&#xff08;天津&#xff09;自由贸易试验区政策与产业创新发展局、网络空间治理与数字经济法治&#xff08;长三角&…
阅读更多...
称球问题之十二个球有一个次品不知轻重三次称出

称球问题之十二个球有一个次品不知轻重三次称出

使用穷举法来解&#xff1a; 12个球分A1234、B1234、C1234 1、A1234B1234→坏在C1234 1.1、A123C123→坏在C4 1.2、A123<C123→坏在C123&#xff0c;且知坏的重 1.2.1、C1C2→坏在C3 1.2.2、C1<C2→坏在C2 1.2.3、C1>C2→坏在C1 1.3、A123>C123→坏在C123&…
阅读更多...
24位AD转换器16路模拟信号转RS-485/232，数据采集A/D转换模块 YL29

24位AD转换器16路模拟信号转RS-485/232，数据采集A/D转换模块 YL29

特点&#xff1a; ● 16路模拟信号采集&#xff0c;隔离转换 RS-485/232输出 ● 采用24位AD转换器&#xff0c;测量精度优于0.05% ● 通过RS-485/232接口可以程控校准模块精度 ● 信号输入 / 输出之间隔离耐压3000VDC ● 宽电源供电范围&#xff1a;8 ~ 32VDC ● 可靠性高…
阅读更多...
申请流量卡时，运营商到底审核什么？

申请流量卡时，运营商到底审核什么？

当我们在网上申请流量卡时&#xff0c;会有一个运营商审核的过程&#xff0c;只有运营商审核通过后才会发卡&#xff0c;那么问题来了&#xff0c;运营商审核&#xff0c;到底是审核的什么呢&#xff1f; 申请流量卡时&#xff0c;运营商的审核是非常严格的&#xff0c;如果不…
阅读更多...
【开源】基于SpringBoot的天然气工程业务管理系统的设计和实现

【开源】基于SpringBoot的天然气工程业务管理系统的设计和实现

目录 一、摘要1.1 项目介绍1.2 项目录屏 二、功能模块三、使用角色3.1 施工人员3.2 管理员 四、数据库设计4.1 用户表4.2 分公司表4.3 角色表4.4 数据字典表4.5 工程项目表4.6 使用材料表4.7 使用材料领用表4.8 整体E-R图 五、系统展示六、核心代码6.1 查询工程项目6.2 工程物资…
阅读更多...
TypeError: lineplot() takes from 0 to 1 positional arguments but 2 were given

TypeError: lineplot() takes from 0 to 1 positional arguments but 2 were given

使用pyplot生成图形 from matplotlib import pyplot as plt import pandas as pd import seaborn as sns import numpy as np# 设置中文字体 plt.rcParams[font.sans-serif] [SimHei] sns.set_style({font.sans-serif:[simhei, Arial]})# 加载数据 hr pd.read_csv(data/hr.c…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023