Linux内核中RPS/RFS代码分析

netdev_rx_queue表示对应的接收队列，很多网卡硬件上已经支持多个队列，此时就会有多个netdev_rx_queue队列，这个结构是挂在net_device，初始化接收队列的函数：netif_alloc_rx_queues

netif_alloc_rx_queues

1. static int netif_alloc_rx_queues(struct net_device *dev)
   
2. {
   
3. /*获取接收队列的个数*/
   
4.     unsigned int i, count = netdev_extended(dev)->rps_data.num_rx_queues;
   
5.     struct netdev_rx_queue *rx;
   
6. 
   
7.     BUG_ON(count < 1);
   
8.  /*分配netdev_rx_queue 空间*/
   
9.     rx = kcalloc(count, sizeof(struct netdev_rx_queue), GFP_KERNEL);
   
10.     if (!rx) {
    
11.         pr_err("netdev: Unable to allocate %u rx queues.\n", count);
    
12.         return -ENOMEM;
    
13.     }
    
14.  /* netdev_rx_queue 和net_device关联起来。*/
    
15.     netdev_extended(dev)->rps_data._rx = rx;
    
16.  /*对netdev_rx_queue 中net_device进行赋值操作*/
    
17.     for (i = 0; i < count; i++)
    
18.         rx[i].dev = dev;
    
19.     return 0;
    
20. }
    
21. 
    
22.     struct netdev_rx_queue {
    
23.     /*保存当前队列的rps map*/
    
24.     struct rps_map *rps_map;
    
25. /* //每个设备的队列保存了一个rps_dev_flow_table */    
    
26. struct rps_dev_flow_table *rps_flow_table;
    
27. //对应的kobject
    
28.     struct kobject kobj;
    
29. /*所属的net_device*/
    
30.     struct net_device *dev;
    
31. } ____cacheline_aligned_in_smp;
    
32. 
    
33. 
    
34. struct rps_map {
    
35. /*CPU的个数，也就是CPU数组的个数*/
    
36.     unsigned int len;
    
37.     struct rcu_head rcu;
    
38. /*保存了CPU的ID*/
    
39.     u16 cpus[0];
    
40. };

get_rps_cpu

1. static int get_rps_cpu(struct net_device *dev, struct sk_buff *skb,
   
2.          struct rps_dev_flow **rflowp)
   
3. {
   
4.     struct ipv6hdr *ip6;
   
5.     struct iphdr *ip;
   
6.     struct netdev_rx_queue *rxqueue;
   
7.     struct rps_map *map;
   
8.     struct rps_dev_flow_table *flow_table;
   
9.     struct rps_sock_flow_table *sock_flow_table;
   
10.     struct netdev_rps_info *rpinfo = &netdev_extended(dev)->rps_data;
    
11.     int cpu = -1;
    
12.     int tcpu;
    
13.     u8 ip_proto;
    
14.     u32 addr1, addr2, ports, ihl;
    
15. 
    
16.     rcu_read_lock();
    
17. 
    
18.     if (skb_rx_queue_recorded(skb)) {
    
19. /*获取设备对应的rx队列。*/
    
20.         u16 index = skb_get_rx_queue(skb);
    
21.         if (unlikely(index >= rpinfo->num_rx_queues)) {
    
22.             WARN_ONCE(rpinfo->num_rx_queues > 1, "%s received packet "
    
23.                 "on queue %u, but number of RX queues is %u\n",
    
24.                 dev->name, index, rpinfo->num_rx_queues);
    
25.             goto done;
    
26.         }
    
27.         rxqueue = rpinfo->_rx + index;
    
28.     } else
    
29.         rxqueue = rpinfo->_rx;
    
30. 
    
31.     if (!rxqueue->rps_map && !rxqueue->rps_flow_table)
    
32.         goto done;
    
33. 
    
34.     if (skb->rxhash) //如果硬件已经计算过，则直接跳过，不需要计算HASH值
    
35.         goto got_hash; /* Skip hash computation on packet header */
    
36. 
    
37.     switch (skb->protocol) { /*根据不同的IP协议获取源IP和目的IP*/
    
38.     case __constant_htons(ETH_P_IP):
    
39.         if (!pskb_may_pull(skb, sizeof(*ip)))
    
40.             goto done;
    
41. 
    
42.         ip = (struct iphdr *) skb->data;
    
43.         ip_proto = ip->protocol;
    
44.         addr1 = ip->saddr;
    
45.         addr2 = ip->daddr;
    
46.         ihl = ip->ihl;
    
47.         break;
    
48.     case __constant_htons(ETH_P_IPV6):
    
49.         if (!pskb_may_pull(skb, sizeof(*ip6)))
    
50.             goto done;
    
51. 
    
52.         ip6 = (struct ipv6hdr *) skb->data;
    
53.         ip_proto = ip6->nexthdr;
    
54.         addr1 = ip6->saddr.s6_addr32[3];
    
55.         addr2 = ip6->daddr.s6_addr32[3];
    
56.         ihl = (40 >> 2);
    
57.         break;
    
58.     default:
    
59.         goto done;
    
60.     }
    
61.     ports = 0;
    
62.     switch (ip_proto) {
    
63.     case IPPROTO_TCP:
    
64.     case IPPROTO_UDP:
    
65.     case IPPROTO_DCCP:
    
66.     case IPPROTO_ESP:
    
67.     case IPPROTO_AH:
    
68.     case IPPROTO_SCTP:
    
69.     case IPPROTO_UDPLITE:
    
70.         if (pskb_may_pull(skb, (ihl * 4) + 4))
    
71.             ports = *((u32 *) (skb->data + (ihl * 4))); /*获取四层协议的端口号，tcp头的前4个字节就是源和目的端口，因此这里跳过ip头得到tcp头的前4个字节*/
    
72.         break;
    
73. 
    
74.     default:
    
75.         break;
    
76.     }
    
77.  /*根据获取到的SIP和DIP，PORT计算HSAH值，*/
    
78.     skb->rxhash = jhash_3words(addr1, addr2, ports, hashrnd) >> 16;
    
79.     if (!skb->rxhash)
    
80.         skb->rxhash = 1;
    
81. 
    
82. got_hash:
    
83. /* rps_sock_flow_table和rps_dev_flow_table 是为了解决RFS而添加的两张表，rps_sock_flow_table是一个全局的hash表，这个表针对socket的，映射了socket对应的cpu，这里的cpu就是应用层期待软中断所在的cpu ，rps_dev_flow_table,这个是针对设备的，每个设备队列都含有一个rps_dev_flow_table(这个表主要是保存了上次处理相同链接上的skb所在的cpu),这个hash表中每一个元素包含了一个cpu id，一个tail queue的计数器*/
    
84.     flow_table = rcu_dereference(rxqueue->rps_flow_table);
    
85.     sock_flow_table = rcu_dereference(rps_sock_flow_table);
    
86.     if (flow_table && sock_flow_table) {
    
87.         u16 next_cpu;
    
88.         struct rps_dev_flow *rflow;
    
89. 
    
90.         rflow = &flow_table->flows[skb->rxhash & flow_table->mask];
    
91.         tcpu = rflow->cpu;
    
92. 
    
93.         next_cpu = sock_flow_table->ents[skb->rxhash &
    
94.          sock_flow_table->mask];
    
95. 
    
96.         /*首先会得到两个flow table，一个是sock_flow_table,另一个是设备的rps_flow_table(skb对应的设备队列中对应的flow table)，这里的逻辑是这样子的取出来两个cpu，一个是根据rps计算数据包前一次被调度过的cpu(tcpu)，一个是应用程序期望的cpu(next_cpu)，然后比较这两个值，如果 1 tcpu未设置(等于RPS_NO_CPU） 2 tcpu是离线的 3 tcpu的input_queue_head大于rps_flow_table中的last_qtail 的话就调度这个skb到next_cpu.而这里第三点input_queue_head大于rps_flow_table则说明在当前的dev flow table中的数据包已经发送完毕，否则的话为了避免乱序就还是继续使用tcpu
    
97.          * If the desired CPU (where last recvmsg was done) is
    
98.          * different from current CPU (one in the rx-queue flow
    
99.          * table entry), switch if one of the following holds:
    
100.          * - Current CPU is unset (equal to RPS_NO_CPU).
     
101.          * - Current CPU is offline.
     
102.          * - The current CPU's queue tail has advanced beyond the
     
103.          * last packet that was enqueued using this table entry.
     
104.          * This guarantees that all previous packets for the flow
     
105.          * have been dequeued, thus preserving in order delivery.
     
106.          */
     
107.         if (unlikely(tcpu != next_cpu) &&
     
108.          (tcpu == RPS_NO_CPU || !cpu_online(tcpu) ||
     
109.          ((int)(per_cpu(softnet_data, tcpu).input_queue_head -
     
110.          rflow->last_qtail)) >= 0)) {
     
111.             tcpu = rflow->cpu = next_cpu;
     
112.             if (tcpu != RPS_NO_CPU)
     
113.                 rflow->last_qtail = per_cpu(softnet_data,
     
114.                  tcpu).input_queue_head;
     
115.         }
     
116.         if (tcpu != RPS_NO_CPU && cpu_online(tcpu)) {
     
117.             *rflowp = rflow;
     
118.  /*设置返回的CPU*/
     
119.             cpu = tcpu;
     
120.             goto done;
     
121.         }
     
122.     }
     
123. /*当第一次进来时tcpu是RPS_NO_CPU,并且next_cpu也是RPS_NO_CPU，此时会导致跳过rfs处理，而是直接使用rps的处理, */
     
124.     map = rcu_dereference(rxqueue->rps_map);
     
125.     if (map) {
     
126.         tcpu = map->cpus[((u32) (skb->rxhash * map->len)) >> 16];
     
127. /*如果cpu是online的，则返回计算出的这个cpu */
     
128.         if (cpu_online(tcpu)) {
     
129.             cpu = tcpu;
     
130.             goto done;
     
131.         }
     
132.     }
     
133. 
     
134. done:
     
135.     rcu_read_unlock();
     
136.     return cpu;
     
137. }
     
138. 
     
139. /*将skb挂在到对应cpu的input queue上的, enqueue_to_backlog接受一个skb和cpu为参数，通过cpu来判断skb如何处理。要么加入所属的input_pkt_queue中，要么schecule 软中断*/

enqueue_to_backlog

1. static int enqueue_to_backlog(struct sk_buff *skb, int cpu,
   
2.              unsigned int *qtail)
   
3. {
   
4.     struct softnet_data *queue;
   
5.     unsigned long flags;
   
6.    /*根据传递过来的CPU，获取softnet_data结构体*/
   
7.     queue = &per_cpu(softnet_data, cpu);
   
8. 
   
9.     local_irq_save(flags);
   
10.     __get_cpu_var(netdev_rx_stat).total++;
    
11. 
    
12.     spin_lock(&queue->input_pkt_queue.lock);
    
13.     if (queue->input_pkt_queue.qlen <= netdev_max_backlog) {
    
14.         if (queue->input_pkt_queue.qlen) {
    
15. enqueue:/*将数据包添加到input_pkt_queue队列中*/
    
16.             __skb_queue_tail(&queue->input_pkt_queue, skb);
    
17.             *qtail = queue->input_queue_head +
    
18.                  queue->input_pkt_queue.qlen;
    
19. 
    
20.             spin_unlock_irqrestore(&queue->input_pkt_queue.lock,
    
21.              flags);
    
22.             return NET_RX_SUCCESS;
    
23.         }
    
24. 
    
25.         /* Schedule NAPI for backlog device 可以调度软中断*/
    
26.         if (napi_schedule_prep(&queue->backlog)) {
    
27.             if (cpu != smp_processor_id()) {/*判断该SKB是否该CPU处理*/
    
28.                 struct rps_remote_softirq_cpus *rcpus =
    
29.                  &__get_cpu_var(rps_remote_softirq_cpus);
    
30. 
    
31.                 cpu_set(cpu, rcpus->mask[rcpus->select]);
    
32.                 __raise_softirq_irqoff(NET_RX_SOFTIRQ);
    
33.             } else
    
34. /*应该当前cpu处理，则直接schedule 软中断，这里可以看到传递进去的是backlog */
    
35.                 ____napi_schedule(queue, &queue->backlog);
    
36.         }
    
37.         goto enqueue;
    
38.     }
    
39. 
    
40.     spin_unlock(&queue->input_pkt_queue.lock);
    
41. 
    
42.     __get_cpu_var(netdev_rx_stat).dropped++;
    
43.     local_irq_restore(flags);
    
44. 
    
45.     kfree_skb(skb);
    
46.     return NET_RX_DROP;
    
47. }

enqueue_to_backlog

1. static int enqueue_to_backlog(struct sk_buff *skb, int cpu,
   
2.              unsigned int *qtail)
   
3. {
   
4.     struct softnet_data *queue;
   
5.     unsigned long flags;
   
6.    /*根据传递过来的CPU，获取softnet_data结构体*/
   
7.     queue = &per_cpu(softnet_data, cpu);
   
8. 
   
9.     local_irq_save(flags);
   
10.     __get_cpu_var(netdev_rx_stat).total++;
    
11. 
    
12.     spin_lock(&queue->input_pkt_queue.lock);
    
13.     if (queue->input_pkt_queue.qlen <= netdev_max_backlog) {
    
14.         if (queue->input_pkt_queue.qlen) {
    
15. enqueue:/*将数据包添加到input_pkt_queue队列中*/
    
16.             __skb_queue_tail(&queue->input_pkt_queue, skb);
    
17.             *qtail = queue->input_queue_head +
    
18.                  queue->input_pkt_queue.qlen;
    
19. 
    
20.             spin_unlock_irqrestore(&queue->input_pkt_queue.lock,
    
21.              flags);
    
22.             return NET_RX_SUCCESS;
    
23.         }
    
24. 
    
25.         /* Schedule NAPI for backlog device 可以调度软中断*/
    
26.         if (napi_schedule_prep(&queue->backlog)) {
    
27.             if (cpu != smp_processor_id()) {/*判断该SKB是否该CPU处理*/
    
28.                 struct rps_remote_softirq_cpus *rcpus =
    
29.                  &__get_cpu_var(rps_remote_softirq_cpus);
    
30. 
    
31.                 cpu_set(cpu, rcpus->mask[rcpus->select]);
    
32.                 __raise_softirq_irqoff(NET_RX_SOFTIRQ);
    
33.             } else
    
34. /*应该当前cpu处理，则直接schedule 软中断，这里可以看到传递进去的是backlog */
    
35.                 ____napi_schedule(queue, &queue->backlog);
    
36.         }
    
37.         goto enqueue;
    
38.     }
    
39. 
    
40.     spin_unlock(&queue->input_pkt_queue.lock);
    
41. 
    
42.     __get_cpu_var(netdev_rx_stat).dropped++;
    
43.     local_irq_restore(flags);
    
44. 
    
45.     kfree_skb(skb);
    
46.     return NET_RX_DROP;
    
47. }

inet_recvmsg

1. int inet_recvmsg(struct kiocb *iocb, struct socket *sock, struct msghdr *msg,
   
2.          size_t size, int flags)
   
3. {
   
4.     struct sock *sk = sock->sk;
   
5.     int addr_len = 0;
   
6.     int err;
   
7. 
   
8.     inet_rps_record_flow(sk);//设置HASH表
   
9. 
   
10.     err = sk->sk_prot->recvmsg(iocb, sk, msg, size, flags & MSG_DONTWAIT,
    
11.                  flags & ~MSG_DONTWAIT, &addr_len);
    
12.     if (err >= 0)
    
13.         msg->msg_namelen = addr_len;
    
14.     return err;
    
15. }

这个函数主要是得到全局的rps_sock_flow_table，然后调用rps_record_sock_flow来对rps_sock_flow_table进行设置，这里会将socket的sk_rxhash传递进去当作hash的索引，而这个sk_rxhash其实就是skb里面的rxhash，skb的rxhash就是rps中设置的hash值，这个值是根据四元组进行hash的。这里用这个当索引一个是为了相同的socket都能落入一个index。而且下面的软中断上下文也比较容易存取这个hash表

inet_rps_record_flow

点击(此处)折叠或打开

1. static inline void inet_rps_record_flow(struct sock *sk)
   
2. {
   
3.     struct rps_sock_flow_table *sock_flow_table;
   
4. 
   
5.     rcu_read_lock();
   
6.     sock_flow_table = rcu_dereference(rps_sock_flow_table);
   
7.     rps_record_sock_flow(sock_flow_table, inet_sk_rxhash(sk));
   
8.     rcu_read_unlock();
   
9. }

rps_record_sock_flow

点击(此处)折叠或打开

1. static inline void rps_record_sock_flow(struct rps_sock_flow_table *table,
   
2.                     u32 hash)
   
3. {
   
4.     if (table && hash) {
   
5. /*获取索引*/
   
6.         unsigned int cpu, index = hash & table->mask;
   
7. 
   
8.         /* We only give a hint, preemption can change cpu under us 获取CPU */
   
9.         cpu = raw_smp_processor_id();
   
10.  /*保存对应的cpu,如果等于当前cpu，则说明已经设置过了*/
    
11.         if (table->ents[index] != cpu)
    
12.             table->ents[index] = cpu;
    
13.     }
    
14. }

    图：内核代码流程