本文共 9830 字，大约阅读时间需要 32 分钟。

作者：lwyang?

内核版本：Linux-4.20.8

每次一个以太网帧到达时，都使用一个IRQ来通知内核。这里暗含着“快”和“慢”的概念。 对低速设备来说，在下一个分组到达之前， IRQ的处理通常已经结束。由于下一个分组也通过IRQ通知，如果前一个分组的IRQ尚未处理完成，则会导致问题，高速设备通常就是这样。现代以太网卡的运作高达10 000 Mbit/s，如果使用旧式方法来驱动此类设备，将造成所谓的“中断风暴”。如果在分组等待处理时接收到新的IRQ，内核不会收到新的信息：在分组进入处理过程之前，内核是可以接收IRQ的，在分组的处理结束后，内核也可以接收IRQ，这些不过是“旧闻”而已。为解决该问题，NAPI使用了IRQ和轮询的组合

假定某个网络适配器此前没有分组到达，但从现在开始，分组将以高频率频繁到达。这就是NAPI设备的情况，如下所述。

1. 第一个分组将导致网络适配器发出IRQ。为防止进一步的分组导致发出更多的IRQ，驱动程序
   会关闭该适配器的Rx IRQ。并将该适配器放置到一个轮询表上。
2. 只要适配器上还有分组需要处理，内核就一直对轮询表上的设备进行轮询。
3. 重新启用Rx中断。

如果在新的分组到达时，旧的分组仍然处于处理过程中，工作不会因额外的中断而减速。虽然对设备驱动程序（和一般意义上的内核代码）来说轮询通常是一个很差的方法，但在这里该方法没有什么不利之处：在没有分组还需要处理时，将停止轮询，设备将回复到通常的IRQ驱动的运行方式。在没有中断支持的情况下，轮询空的接收队列将不必要地浪费时间，但NAPI并非如此

NAPI的另一个优点是可以高效地丢弃分组。如果内核确信因为有很多其他工作需要处理，而导致无法处理任何新的分组，那么网络适配器可以直接丢弃分组，无须复制到内核。

只有设备满足如下两个条件时，才能实现NAPI方法。

1. 设备必须能够保留多个接收的分组，例如保存到DMA环形缓冲区中。下文将该缓冲区称为Rx
   缓冲区。
2. 该设备必须能够禁用用于分组接收的IRQ。而且，发送分组或其他可能通过IRQ进行的操作，
   都仍然必须是启用的

NAPI机制和循环轮询表概览:

如果一个分组到达一个空的Rx缓冲区，则将相应的设备置于轮询表中。由于链表本身的性质，轮询表可以包含多个设备。内核以循环方式处理链表上的所有设备：内核依次轮询各个设备，如果已经花费了一定的时间来处理某个设备，则选择下一个设备进行处理。此外，某个设备都带有一个相对权重，表示与轮询表中其他设备相比，该设备的相对重要性。较快的设备权重较大，较慢的设备权重较小。由于权重指定了在一个轮询的循环中处理多少分组，这确保了内核将更多地注意速度较快的设备

NAPI是中断机制与轮询的混合体，当一批数据包中第一个数据包到达网络设备，会以硬中断的方式通知系统，在硬中断例程，系统将该设备添加到CPU的设备轮询队列，并关闭中断，同时激活数据包输入软中断，由软中断例程遍历轮询队列中的网络设备，从中读取数据包。

相关接口层的初始化：net_dev_init

static int __init net_dev_init(void){
   	int i, rc = -ENOMEM;	BUG_ON(!dev_boot_phase);	// 注册记录相关统计信息的proc文件系统	// proc/net/dev : 用于显示网络设备收发包统计信息	// proc/net/softnet_stat : 显示每个CPU的softnet_stat 统计信息	// proc/net/ptype : 显示注册的协议处理函数	if (dev_proc_init())		goto out;			//初始化kobject相关信息，为创建sys文件系统	if (netdev_kobject_init())		goto out;	//初始化 ptype_all链表和ptype_base链表	INIT_LIST_HEAD(&ptype_all);	for (i = 0; i < PTYPE_HASH_SIZE; i++)		INIT_LIST_HEAD(&ptype_base[i]);	//初始化offload_base链表	INIT_LIST_HEAD(&offload_base);	//注册跟网络命名空间相关的信息	//INIT_LIST_HEAD(&net->dev_base_head); 初始化dev_base_head链表	//net->dev_name_head 初始化dev_name_head 散列表，用来根据网络设备名获取网络设备	//net->dev_index_head 初始化dev_index_head 散列表，用来根据设备索引号获取网络设备	if (register_pernet_subsys(&netdev_net_ops))		goto out;	/*	 *	Initialise the packet receive queues.	 */	//初始化每个CPU的softnet_data ，包括发送数据包的等待释放队列，轮询函数等	for_each_possible_cpu(i) {
   		struct work_struct *flush = per_cpu_ptr(&flush_works, i);		struct softnet_data *sd = &per_cpu(softnet_data, i);		INIT_WORK(flush, flush_backlog);		//非napi使用此input_pkt_queue		skb_queue_head_init(&sd->input_pkt_queue);		skb_queue_head_init(&sd->process_queue);#ifdef CONFIG_XFRM_OFFLOAD		skb_queue_head_init(&sd->xfrm_backlog);#endif		INIT_LIST_HEAD(&sd->poll_list);		sd->output_queue_tailp = &sd->output_queue;#ifdef CONFIG_RPS		sd->csd.func = rps_trigger_softirq;		sd->csd.info = sd;		sd->cpu = i;#endif		init_gro_hash(&sd->backlog);		//初始化napi_struct结构		sd->backlog.poll = process_backlog;		sd->backlog.weight = weight_p;	}	dev_boot_phase = 0;	/* The loopback device is special if any other network devices	 * is present in a network namespace the loopback device must	 * be present. Since we now dynamically allocate and free the	 * loopback device ensure this invariant is maintained by	 * keeping the loopback device as the first device on the	 * list of network devices.  Ensuring the loopback devices	 * is the first device that appears and the last network device	 * that disappears.	 */	if (register_pernet_device(&loopback_net_ops))		goto out;	if (register_pernet_device(&default_device_ops))		goto out;	//注册网络报文输出/输入软中断处理例程	open_softirq(NET_TX_SOFTIRQ, net_tx_action);	open_softirq(NET_RX_SOFTIRQ, net_rx_action);	rc = cpuhp_setup_state_nocalls(CPUHP_NET_DEV_DEAD, "net/dev:dead",				       NULL, dev_cpu_dead);	WARN_ON(rc < 0);	rc = 0;out:	return rc;}subsys_initcall(net_dev_init);

softnet_data结构描述了与网络软中断处理相关的报文输入和输出队列，每个CPU有一个单独的softnet_data，因此在操作该结构的成员时不必加锁

中断上半部

//随意举例一个驱动处理函数static int vortex_rx(struct net_device *dev){
   	...	//分配skb，包含skb->dev = dev	skb = netdev_alloc_skb(dev, pkt_len + 5);	if (skb != NULL) {
   		...		skb_reserve(skb, 2);	/* Align IP on 16 byte boundaries */		skb->protocol = eth_type_trans(skb, dev);		netif_rx(skb);		dev->stats.rx_packets++;		...	}	...}

在进入netif_rx之前，已经设置好了skb->dev

/** * eth_type_trans - determine the packet's protocol ID. * @skb: received socket data * @dev: receiving network device * * The rule here is that we * assume 802.3 if the type field is short enough to be a length. * This is normal practice and works for any 'now in use' protocol. */__be16 eth_type_trans(struct sk_buff *skb, struct net_device *dev){
   	unsigned short _service_access_point;	const unsigned short *sap;	const struct ethhdr *eth;	//记录收包的设备	skb->dev = dev;		//设置skb->mac_header,即以太网头部的偏移量（skb->mac_header = skb->data - skb->head）	//只有这里设置了skb->mac_header，才能使用eth_hdr函数（skb->head + skb->mac_header）获取以太网头部	skb_reset_mac_header(skb);	//获取以太网头部	eth = (struct ethhdr *)skb->data;	//将data指针指向ip层头部（没有vlan的情况下）	//在有vlan的情况下data指针 是指向vlan tag的tci信息，下节中会深入分析这一做法	skb_pull_inline(skb, ETH_HLEN);	//确定目的mac地址是单播还是广播，设置skb->pkt_type的类型	if (unlikely(is_multicast_ether_addr_64bits(eth->h_dest))) {
   		if (ether_addr_equal_64bits(eth->h_dest, dev->broadcast))			skb->pkt_type = PACKET_BROADCAST;		else			skb->pkt_type = PACKET_MULTICAST;	}	//目的mac地址不是本机	else if (unlikely(!ether_addr_equal_64bits(eth->h_dest,						   dev->dev_addr)))		skb->pkt_type = PACKET_OTHERHOST;	/*	 * Some variants of DSA tagging don't have an ethertype field	 * at all, so we check here whether one of those tagging	 * variants has been configured on the receiving interface,__netif_receive_skb	 * and if so, set skb->protocol without looking at the packet.	 */	if (unlikely(netdev_uses_dsa(dev)))		return htons(ETH_P_XDSA);	//当h_proto > 1536	//ETH_P_IP       0x0800  iP_rcv	//ETH_P_ARP      0x0806  arp_rcv	//ETH_P_PPP_SES  0x8864  pppoe_rcv 	if (likely(eth_proto_is_802_3(eth->h_proto)))		return eth->h_proto;	/*	 *      This is a magic hack to spot IPX packets. Older Novell breaks	 *      the protocol design and runs IPX over 802.3 without an 802.2 LLC	 *      layer. We look for FFFF which isn't a used 802.2 SSAP/DSAP. This	 *      won't work for fault tolerant netware but does for the rest.	 */	 //IPX数据包没有802.2标准的LLC层,0xFFFF为其标志位	sap = skb_header_pointer(skb, 0, sizeof(*sap), &_service_access_point);	if (sap && *sap == 0xFFFF)		return htons(ETH_P_802_3);	/*	 *      Real 802.2 LLC	 */	 //802.2标准的LLC协议	return htons(ETH_P_802_2);}

Ethernat的地址其实就是Mac地址。所以长度是6byte。其中有一位为multicast bit位。格式如下

以及 skb的data指针已经指向ip层，设置好了skb->pkt_type，skb->protocol，skb->mac_header（使用eth_hdr函数来获取以太网头部的前提）

接下来会调用netif_rx

int netif_rx(struct sk_buff *skb){
   	trace_netif_rx_entry(skb);	return netif_rx_internal(skb);}static int netif_rx_internal(struct sk_buff *skb){
   	int ret;	net_timestamp_check(netdev_tstamp_prequeue, skb);	...	if (static_key_false(&rps_needed)) {
   		...	}	else	{
   		unsigned int qtail;		ret = enqueue_to_backlog(skb, get_cpu(), &qtail);		put_cpu();	}	return ret;}

netif_rx()调用enqueue_to_backlog()来处理

static int enqueue_to_backlog(struct sk_buff *skb, int cpu, unsigned int *qtail){
   	struct softnet_data *sd;	unsigned long flags;	unsigned int qlen;	sd = &per_cpu(softnet_data, cpu);	local_irq_save(flags);	...	//获取input_pkt_queue长度	qlen = skb_queue_len(&sd->input_pkt_queue);	if (qlen <= netdev_max_backlog && !skb_flow_limit(skb, qlen)) {
   			//如果接收队列sd->input_pkt_queue不为空，说明已经有软中断在处理数据包了，		//则不需要再次触发软中断，直接将数据包添加到接收队列尾部即可		if (qlen) {
   enqueue:			__skb_queue_tail(&sd->input_pkt_queue, skb);			input_queue_tail_incr_save(sd, qtail);			rps_unlock(sd);			local_irq_restore(flags);			return NET_RX_SUCCESS;		}		/* Schedule NAPI for backlog device		 * We can use non atomic operation since we own the queue lock		 */		//如果接收队列sd->input_pkt_queue为空，说明当前没有软中断在处理数据包，		//则把虚拟设备backlog添加到sd->poll_list中以便进行轮询，最后设置NET_RX_SOFTIRQ标志触发软中断。		if (!__test_and_set_bit(NAPI_STATE_SCHED, &sd->backlog.state)) {
   			if (!rps_ipi_queued(sd))				____napi_schedule(sd, &sd->backlog);		}		goto enqueue;	}//如果接收队列sd->input_pkt_queue满了，则直接丢弃数据包drop:	sd->dropped++;	rps_unlock(sd);	local_irq_restore(flags);	atomic_long_inc(&skb->dev->rx_dropped);	kfree_skb(skb);	return NET_RX_DROP;}

软中断(NET_RX_SOFTIRQ)的处理函数net_rx_action()

中断下半部

net_rx_action()为网络输入软中断的处理例程，当网络设备有数据输入，非NAPI和NAPI的网络设备驱动程序都会激活网络输入软中断进行处理

static __latent_entropy void net_rx_action(struct softirq_action *h){
   	//获取CPU的softnet_data 	struct softnet_data *sd = this_cpu_ptr(&softnet_data);	//一次中断处理的最长时间	unsigned long time_limit = jiffies +		usecs_to_jiffies(netdev_budget_usecs);	//获取本次软中断的接受报文配额	int budget = netdev_budget;	LIST_HEAD(list);	LIST_HEAD(repoll);	local_irq_disable();	list_splice_init(&sd->poll_list, &list);	local_irq_enable();	//遍历网络设备轮询队列上的网络设备，轮询接受这些网络设备上的报文	for (;;) {
   		struct napi_struct *n;		if (list_empty(&list)) {
   			if (!sd_has_rps_ipi_waiting(sd) && list_empty(&repoll))				goto out;			break;		}		//获取链表上的第一个napi_struct实例		n = list_first_entry(&list, struct napi_struct, poll_list);		//调用napi_struct的poll方法,默认为process_backlog()		//在之前的net_dev_init中  sd->backlog.poll = process_backlog;		budget -= napi_poll(n, &repoll);		...	}	...out:	__kfree_skb_flush();}

当调用napi_struct的poll()来处理数据包时，本地中断是开启的，这意味着新的数据包可以继续添加到输入队列中

接下来会调用napi_struct的poll方法，如果网卡驱动不支持NAPI，则默认的napi_struct->poll()函数为process_backlog

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;	...		//每次处理的最大数据包数	napi->weight = dev_rx_weight;	while (again) {
   		struct sk_buff *skb;		//获取当前待处理报文，获取失败则说明队列为空，		while ((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;		}		...	return work;}

接下来就调用__netif_receive_skb进入桥进行转发或上交到协议栈处理

转载地址：http://ryhpi.baihongyu.com/