「十八」数据包的转发 dev_forward_skb源码分析

int dev_forward_skb(struct net_device *dev, struct sk_buff *skb)
{
    return __dev_forward_skb(dev, skb) ?: netif_rx_internal(skb);
}

int __dev_forward_skb(struct net_device *dev, struct sk_buff *skb)
{
    int ret = ____dev_forward_skb(dev, skb);

    if (likely(!ret)) {
        skb->protocol = eth_type_trans(skb, dev);
        skb_postpull_rcsum(skb, eth_hdr(skb), ETH_HLEN);
    }

    return ret;
}

static __always_inline int ____dev_forward_skb(struct net_device *dev,
                           struct sk_buff *skb)
{
    if (skb_orphan_frags(skb, GFP_ATOMIC) ||
        unlikely(!is_skb_forwardable(dev, skb))) {
        atomic_long_inc(&dev->rx_dropped);
        kfree_skb(skb);
        return NET_RX_DROP;
    }

    skb_scrub_packet(skb, true);
    skb->priority = 0;
    return 0;
}

void skb_scrub_packet(struct sk_buff *skb, bool xnet)
{
    //记录时间戳
    skb->tstamp = 0;
    //表示数据包发给谁，PACKET_HOST表示发送给本机
    skb->pkt_type = PACKET_HOST;
    //数据包到达接口的索引号
    skb->skb_iif = 0;
    //赋值为0表示不允许本地分片（local fragmentation）
    skb->ignore_df = 0;
    //清除skb的destination entry，即路由项
    skb_dst_drop(skb);
    //与xfrm安全框架有关，清除skb的security path
    secpath_reset(skb);
    //一下两个函数调用都与netfilter packet trace flag，这里忽略详细分析
    nf_reset(skb);
    nf_reset_trace(skb);
    //如果不是跨网络命名空间传递数据包，那么到这里清除任务就结束了，但是从上文可以知道，传入的是true，所以还得继续
    if (!xnet)
        return;
    //将skb的ipvs_property设为0，该属性与ipvs（Linux Virtual Server）有关
    ipvs_reset(skb);
    skb_orphan(skb);
    //清除普通包标记
    skb->mark = 0;
}

static inline void skb_orphan(struct sk_buff *skb)
{
    if (skb->destructor) {
        skb->destructor(skb);
        skb->destructor = NULL;
        skb->sk     = NULL;
    } else {
        BUG_ON(skb->sk);
    }
}

static int netif_rx_internal(struct sk_buff *skb)
{
    int ret;
    //设置skb的tstamp值，此值是记录接收包的时间，@tstamp: Time we arrived/left
    net_timestamp_check(netdev_tstamp_prequeue, skb);
    
    trace_netif_rx(skb);
    
    if (static_key_false(&generic_xdp_needed)) {
        int ret;
        //禁止抢占
        preempt_disable();
        rcu_read_lock();
        ret = do_xdp_generic(rcu_dereference(skb->dev->xdp_prog), skb);
        rcu_read_unlock();
        preempt_enable();

        /* Consider XDP consuming the packet a success from
         * the netdev point of view we do not want to count
         * this as an error.
         */
        if (ret != XDP_PASS)
            return NET_RX_SUCCESS;
    }

//RPS 和 RFS 相关代码；RPS: Receive Packet Steering (接收端包的控制)；RFS: Receive Flow Steering (接收端流的控制)；对于多处理器系统，当使能了RPS（receive packet steering）功能，则会根据负载均衡的原则将数据包平均分配到各个CPU，减少单个CPU的工作负荷[参考7]
#ifdef CONFIG_RPS
    if (static_key_false(&rps_needed)) {
        struct rps_dev_flow voidflow, *rflow = &voidflow;
        int cpu;

        preempt_disable();
        rcu_read_lock();
        
        //选择合适的CPU id
        cpu = get_rps_cpu(skb->dev, skb, &rflow);
        if (cpu < 0)
            cpu = smp_processor_id();
        
        //将skb入队
        ret = enqueue_to_backlog(skb, cpu, &rflow->last_qtail);

        rcu_read_unlock();
        preempt_enable();
    } else
#endif
    {
        unsigned int qtail;
        
        //将skb入队
        ret = enqueue_to_backlog(skb, get_cpu(), &qtail);
        put_cpu();
    }
    return ret;
}

struct softnet_data {
    ...
    struct list_head    poll_list;  //连接所有的轮询设备
    struct sk_buff_head process_queue;  //用于非NAPI设备，因为NAPI设备有自己的队列，处理数据时从该队列取，负责处理
    struct sk_buff_head input_pkt_queue;  //用于非NAPI设备，因为NAPI设备有自己的队列，数据包到来时首先填充到该队列，负责接收
    struct napi_struct  backlog;  //用于非NAPI设备，代表一个虚拟设备供轮询使用，当轮询到该设备时，就会使用以上两个队列
    ...
}

static int enqueue_to_backlog(struct sk_buff *skb, int cpu,
                  unsigned int *qtail)
{
    struct softnet_data *sd;
    unsigned long flags;
    unsigned int qlen;
    //获得指定cpu的softnet_data结构体
    sd = &per_cpu(softnet_data, cpu);
    //关闭中断
    local_irq_save(flags);

    rps_lock(sd);
    //如果skb要到达的dev，在本文中分析的即veth peer是关闭的，则直接跳到drop标签，即丢包
    if (!netif_running(skb->dev))
        goto drop;
    //获取softnet_data中input_pkt_queue队列的长度
    qlen = skb_queue_len(&sd->input_pkt_queue);
    //如果input_pkt_queue队列的长度没有超过设备队列长度最大值，其中skb_flow_limit与上文提到的RPS有关
    if (qlen <= netdev_max_backlog && !skb_flow_limit(skb, qlen)) {
        //如果input_pkt_queue不为空，说明虚拟设备已经得到调度，此时仅仅把数据加入input_pkt_queue队列即可
        if (qlen) {
enqueue:
            //将skb保存到input_pkt_queue队列中
            __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
         */
        //只有qlen是0(表示虚拟设备没有被调度)的时候才执行到这里
        /*
        否则需要调度backlog即虚拟设备，然后再入队。napi_struct实例backlog中的state字段如果标记了NAPI_STATE_SCHED,则表明该设备已经在调度，不需要再次调度
        NAPI_STATE_SCHED：表示设备将在内核的下一次循环时被轮询
        NAPI_STATE_DISABLE：表示轮询已经结束且没有更多的分组等待处理，但设备并没有从poll list移除
        */
        if (!__test_and_set_bit(NAPI_STATE_SCHED, &sd->backlog.state)) {
            //未处于调度状态
            if (!rps_ipi_queued(sd))
                /*
                该函数把设备对应的napi_struct结构（每个设备对应一个napi_struct结构）插入到softnet_data的poll_list链表尾部，然后唤醒软中断，这样在下次软中断得到处理时，中断下半部就会得到处理。
				对于NAPI设备，传入的是自己的napi_struct，而对于非NAPI设备，则传入虚拟设备backlog。
                */
                ____napi_schedule(sd, &sd->backlog);
        }
        goto enqueue;
    }

drop:
    sd->dropped++;
    rps_unlock(sd);

    local_irq_restore(flags);

    atomic_long_inc(&skb->dev->rx_dropped);
    kfree_skb(skb);
    return NET_RX_DROP;
}

static inline void ____napi_schedule(struct softnet_data *sd,
                     struct napi_struct *napi)
{
    //将该设备添加到softnet_data的poll_list队列中
    list_add_tail(&napi->poll_list, &sd->poll_list);
    //唤醒软中断，调用默认或者对应的NAPI设备(veth peer)来收包
    __raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

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;

	/* Check if we have pending ipi, its better to send them now,
	 * not waiting net_rx_action() end.
	 */
	if (sd_has_rps_ipi_waiting(sd)) {
		local_irq_disable();
		net_rps_action_and_irq_enable(sd);
	}

	napi->weight = dev_rx_weight;
	while (again) {
		struct sk_buff *skb;
		/*
			涉及到两个队列process_queue和input_pkt_queue，数据包到来时首先填充input_pkt_queue，
			而在处理时从process_queue中取，根据这个逻辑，首次处理process_queue必定为空，检查input_pkt_queue
			如果input_pkt_queue不为空，则把其中的数据包迁移到process_queue中，然后继续处理，减少锁冲突。
		*/
		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;

		}

		local_irq_disable();
		rps_lock(sd);
		if (skb_queue_empty(&sd->input_pkt_queue)) {
			/*
			 * Inline a custom version of __napi_complete().
			 * only current cpu owns and manipulates this napi,
			 * and NAPI_STATE_SCHED is the only possible flag set
			 * on backlog.
			 * We can use a plain write instead of clear_bit(),
			 * and we dont need an smp_mb() memory barrier.
			 */
			napi->state = 0;
			again = false;
		} else {
			skb_queue_splice_tail_init(&sd->input_pkt_queue,
						   &sd->process_queue);
		}
		rps_unlock(sd);
		local_irq_enable();
	}

	return work;
}