Receive packet steering patch详解

struct netdev_rx_queue {
	// 保存了当前队列的rps map
	struct rps_map *rps_map;
	// 对应的kobject
	struct kobject kobj;
	// 指向第一个rx队列
	struct netdev_rx_queue *first;
	// 引用计数
	atomic_t count;
} ____cacheline_aligned_in_smp;

struct rps_map {
	// cpu的个数，也就是cpus数组的个数
	unsigned int len;
	// RCU锁
	struct rcu_head rcu;
	// 保存了cpu的id.
	u16 cpus[0];
};

static int get_rps_cpu(struct net_device *dev, struct sk_buff *skb)
{
	struct ipv6hdr *ip6;
	struct iphdr *ip;
	struct netdev_rx_queue *rxqueue;
	struct rps_map *map;
	int cpu = -1;
	u8 ip_proto;
	u32 addr1, addr2, ports, ihl;
	// rcu锁
	rcu_read_lock();
	// 取得设备对应的rx 队列
	if (skb_rx_queue_recorded(skb)) {
	..........................................
		rxqueue = dev->_rx + index;
	} else
		rxqueue = dev->_rx;

	if (!rxqueue->rps_map)
		goto done;
	// 如果硬件已经计算，则跳过计算过程
	if (skb->rxhash)
		goto got_hash; /* Skip hash computation on packet header */

	switch (skb->protocol) {
	case __constant_htons(ETH_P_IP):
		if (!pskb_may_pull(skb, sizeof(*ip)))
			goto done;
		// 得到计算hash的几个值
		ip = (struct iphdr *) skb->data;
		ip_proto = ip->protocol;
		// 两个地址
		addr1 = ip->saddr;
		addr2 = ip->daddr;
		// 得到ip头
		ihl = ip->ihl;
		break;
	case __constant_htons(ETH_P_IPV6):
		..........................................
		break;
	default:
		goto done;
	}
	ports = 0;
	switch (ip_proto) {
	case IPPROTO_TCP:
	case IPPROTO_UDP:
	case IPPROTO_DCCP:
	case IPPROTO_ESP:
	case IPPROTO_AH:
	case IPPROTO_SCTP:
	case IPPROTO_UDPLITE:
		if (pskb_may_pull(skb, (ihl * 4) + 4))
		// 我们知道tcp头的前4个字节就是源和目的端口，因此这里跳过ip头得到tcp头的前4个字节
			ports = *((u32 *) (skb->data + (ihl * 4)));
		break;

	default:
		break;
	}
	// 计算hash
	skb->rxhash = jhash_3words(addr1, addr2, ports, hashrnd);
	if (!skb->rxhash)
		skb->rxhash = 1;

got_hash:
	// 通过rcu得到对应rps map
	map = rcu_dereference(rxqueue->rps_map);
	if (map) {
		// 取得对应的cpu
		u16 tcpu = map->cpus[((u64) skb->rxhash * map->len) >> 32];
		// 如果cpu是online的，则返回计算出的这个cpu，否则跳出循环。
		if (cpu_online(tcpu)) {
			cpu = tcpu;
			goto done;
		}
	}

done:
	rcu_read_unlock();
	// 如果上面失败，则返回-1.
	return cpu;
}

/*
 * This structure holds the per-CPU mask of CPUs for which IPIs are scheduled
 * to be sent to kick remote softirq processing.  There are two masks since
 * the sending of IPIs must be done with interrupts enabled.  The select field
 * indicates the current mask that enqueue_backlog uses to schedule IPIs.
 * select is flipped before net_rps_action is called while still under lock,
 * net_rps_action then uses the non-selected mask to send the IPIs and clears
 * it without conflicting with enqueue_backlog operation.
 */
struct rps_remote_softirq_cpus {
	// 对应的cpu掩码
	cpumask_t mask[2];
	// 表示应该使用的数组索引
	int select;
};

static int enqueue_to_backlog(struct sk_buff *skb, int cpu)
{
	struct softnet_data *queue;
	unsigned long flags;
	// 取出传递进来的cpu的softnet-data结构
	queue = &per_cpu(softnet_data, cpu);

	local_irq_save(flags);
	__get_cpu_var(netdev_rx_stat).total++;
	// 自旋锁
	spin_lock(&queue->input_pkt_queue.lock);
	// 如果保存的队列还没到上限
	if (queue->input_pkt_queue.qlen <= netdev_max_backlog) {
	// 如果当前队列的输入队列长度不为空
		if (queue->input_pkt_queue.qlen) {
enqueue:
			// 将数据包加入到input_pkt_queue中,这里会有一个小问题，我们后面再说。
			__skb_queue_tail(&queue->input_pkt_queue, skb);
			spin_unlock_irqrestore(&queue->input_pkt_queue.lock,
				flags);
			return NET_RX_SUCCESS;
		}

		/* Schedule NAPI for backlog device */
		// 如果可以调度软中断
		if (napi_schedule_prep(&queue->backlog)) {
			// 首先判断数据包该不该当前的cpu处理
			if (cpu != smp_processor_id()) {
				// 如果不该，
				struct rps_remote_softirq_cpus *rcpus =
					&__get_cpu_var(rps_remote_softirq_cpus);

				cpu_set(cpu, rcpus->mask[rcpus->select]);
				__raise_softirq_irqoff(NET_RX_SOFTIRQ);
			} else
				// 如果就是应该当前cpu处理，则直接schedule 软中断，这里可以看到传递进去的是backlog
				__napi_schedule(&queue->backlog);
		}
		goto enqueue;
	}

	spin_unlock(&queue->input_pkt_queue.lock);

	__get_cpu_var(netdev_rx_stat).dropped++;
	local_irq_restore(flags);

	kfree_skb(skb);
	return NET_RX_DROP;
}

// 得到对应的rcpus.
rcpus = &__get_cpu_var(rps_remote_softirq_cpus);
	select = rcpus->select;
	// 翻转select，防止和enqueue_backlog冲突
	rcpus->select ^= 1;

	// 打开中断，此时下面的调度才会起作用.
	local_irq_enable();
	// 这个函数里面调度对应的远程cpu的napi.
	net_rps_action(&rcpus->mask[select]);

static void net_rps_action(cpumask_t *mask)
{
	int cpu;

	/* Send pending IPI's to kick RPS processing on remote cpus. */
	// 遍历
	for_each_cpu_mask_nr(cpu, *mask) {
		struct softnet_data *queue = &per_cpu(softnet_data, cpu);
		if (cpu_online(cpu))
			// 到对应的cpu调用csd方法。
			__smp_call_function_single(cpu, &queue->csd, 0);
	}
	// 清理mask
	cpus_clear(*mask);
}

static void trigger_softirq(void *data)
{
	struct softnet_data *queue = data;
	// 调度napi可以看到依旧是backlog 这个napi结构体。
	__napi_schedule(&queue->backlog);
	__get_cpu_var(netdev_rx_stat).received_rps++;
}

int netif_rx(struct sk_buff *skb)
{
	int cpu;

	/* if netpoll wants it, pretend we never saw it */
	if (netpoll_rx(skb))
		return NET_RX_DROP;

	if (!skb->tstamp.tv64)
		net_timestamp(skb);
	// 得到cpu id。
	cpu = get_rps_cpu(skb->dev, skb);
	if (cpu < 0)
		cpu = smp_processor_id();
	// 通过cpu进行队列不同的处理
	return enqueue_to_backlog(skb, cpu);
}

int netif_receive_skb(struct sk_buff *skb)
{
	int cpu;

	cpu = get_rps_cpu(skb->dev, skb);

	if (cpu < 0)
		return __netif_receive_skb(skb);
	else
		return enqueue_to_backlog(skb, cpu);
}