Receive packet steering patch详解

simohayha_cu

Receive packet steering简称rps，是google贡献给linux kernel的一个patch，主要的功能是解决多核情况下，网络协议栈的软中断的负载均衡。这里的负载均衡也就是指能够将软中断均衡的放在不同的cpu核心上运行。

简介在这里：
http://lwn.net/Articles/362339/

linux现在网卡的驱动支持两种模式，一种是NAPI，一种是非NAPI模式，这两种模式的区别，我前面的blog都有介绍，这里就再次简要的介绍下。

在NAPI中，中断收到数据包后调用__napi_schedule调度软中断，然后软中断处理函数中会调用注册的poll回掉函数中调用netif_receive_skb将数据包发送到3层，没有进行任何的软中断负载均衡。

在非NAPI中，中断收到数据包后调用netif_rx，这个函数会将数据包保存到input_pkt_queue，然后调度软中断，这里为了兼容NAPI的驱动，他的poll方法默认是process_backlog，最终这个函数会从input_pkt_queue中取得数据包然后发送到3层。

通过比较我们可以看到，不管是NAPI还是非NAPI的话都无法做到软中断的负载均衡，因为软中断此时都是运行在在硬件中断相应的cpu上。也就是说如果始终是cpu0相应网卡的硬件中断，那么始终都是cpu0在处理软中断，而此时cpu1就被浪费了，因为无法并行的执行多个软中断。

google的这个patch的基本原理是这样的,根据数据包的源地址，目的地址以及目的和源端口(这里它是将两个端口组合成一个4字节的无符数进行计算的，后面会看到)计算出一个hash值，然后根据这个hash值来选择软中断运行的cpu，从上层来看，也就是说将每个连接和cpu绑定，并通过这个hash值，来均衡软中断在多个cpu上。

这个介绍比较简单，我们来看代码是如何实现的。

它这里主要是hook了两个内核的函数，一个是netif_rx主要是针对非NAPI的驱动，一个是netif_receive_skb这个主要是针对NAPI的驱动，这两个函数我前面blog都有介绍过，想了解可以看我前面的blog，现在这里我只介绍打过patch的实现。

在看netif_rx和netif_receive_skb之前，我们先来看这个patch中两个重要的函数get_rps_cpu和enqueue_to_backlog，我们一个个看。

先来看相关的两个数据结构，首先是netdev_rx_queue，它表示对应的接收队列，因为有的网卡可能硬件上就支持多队列的模式，此时对应就会有多个rx队列，这个结构是挂载在net_device中的，也就是每个网络设备最终都会有一个或者多个rx队列。这个结构在sys文件系统中的表示类似这样的/sys/class/net/<device>/queues/rx-<n> 几个队列就是rx-n.

1. struct netdev_rx_queue {
   
2. //保存了当前队列的rps map
   
3.         struct rps_map *rps_map;
   
4. //对应的kobject
   
5.         struct kobject kobj;
   
6. //指向第一个rx队列
   
7.         struct netdev_rx_queue *first;
   
8. //引用计数
   
9.         atomic_t count;
   
10. } ____cacheline_aligned_in_smp;

复制代码

然后就是rps_map，其实这个也就是保存了能够执行数据包的cpu。

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

复制代码

看完上面的结构，我们来看函数的实现。
get_rps_cpu主要是通过传递进来的skb然后来选择这个skb所应该被处理的cpu。它的逻辑很简单，就是通过skb计算hash，然后通过hash从对应的队列的rps_mapping中取得对应的cpu id。

这里有个要注意的就是这个hash值是可以交给硬件网卡去计算的，作者自己说是最好交由硬件去计算这个hash值，因为如果是软件计算的话会导致CPU 缓存不命中，带来一定的性能开销。

还有就是rps_mapping这个值是可以通过sys 文件系统设置的，位置在这里：
/sys/class/net/<device>/queues/rx-<n>/rps_cpus 。

1. static int get_rps_cpu(struct net_device *dev, struct sk_buff *skb)
   
2. {
   
3.         struct ipv6hdr *ip6;
   
4.         struct iphdr *ip;
   
5.         struct netdev_rx_queue *rxqueue;
   
6.         struct rps_map *map;
   
7.         int cpu = -1;
   
8.         u8 ip_proto;
   
9.         u32 addr1, addr2, ports, ihl;
   
10. //rcu锁
    
11.         rcu_read_lock();
    
12. //取得设备对应的rx 队列
    
13.         if (skb_rx_queue_recorded(skb)) {
    
14.         ..........................................
    
15.                 rxqueue = dev->_rx + index;
    
16.         } else
    
17.                 rxqueue = dev->_rx;
    
18. 
    
19.         if (!rxqueue->rps_map)
    
20.                 goto done;
    
21. //如果硬件已经计算，则跳过计算过程
    
22.         if (skb->rxhash)
    
23.                 goto got_hash; /* Skip hash computation on packet header */
    
24. 
    
25.         switch (skb->protocol) {
    
26.         case __constant_htons(ETH_P_IP):
    
27.                 if (!pskb_may_pull(skb, sizeof(*ip)))
    
28.                         goto done;
    
29. //得到计算hash的几个值
    
30.                 ip = (struct iphdr *) skb->data;
    
31.                 ip_proto = ip->protocol;
    
32. //两个地址
    
33.                 addr1 = ip->saddr;
    
34.                 addr2 = ip->daddr;
    
35. //得到ip头
    
36.                 ihl = ip->ihl;
    
37.                 break;
    
38.         case __constant_htons(ETH_P_IPV6):
    
39. ..........................................
    
40.                 break;
    
41.         default:
    
42.                 goto done;
    
43.         }
    
44.         ports = 0;
    
45.         switch (ip_proto) {
    
46.         case IPPROTO_TCP:
    
47.         case IPPROTO_UDP:
    
48.         case IPPROTO_DCCP:
    
49.         case IPPROTO_ESP:
    
50.         case IPPROTO_AH:
    
51.         case IPPROTO_SCTP:
    
52.         case IPPROTO_UDPLITE:
    
53.                 if (pskb_may_pull(skb, (ihl * 4) + 4))
    
54. //我们知道tcp头的前4个字节就是源和目的端口，因此这里跳过ip头得到tcp头的前4个字节
    
55.                         ports = *((u32 *) (skb->data + (ihl * 4)));
    
56.                 break;
    
57. 
    
58.         default:
    
59.                 break;
    
60.         }
    
61. //计算hash
    
62.         skb->rxhash = jhash_3words(addr1, addr2, ports, hashrnd);
    
63.         if (!skb->rxhash)
    
64.                 skb->rxhash = 1;
    
65. 
    
66. got_hash:
    
67. //通过rcu得到对应rps map
    
68.         map = rcu_dereference(rxqueue->rps_map);
    
69.         if (map) {
    
70. //取得对应的cpu
    
71.                 u16 tcpu = map->cpus[((u64) skb->rxhash * map->len) >> 32];
    
72. //如果cpu是online的，则返回计算出的这个cpu，否则跳出循环。
    
73.                 if (cpu_online(tcpu)) {
    
74.                         cpu = tcpu;
    
75.                         goto done;
    
76.                 }
    
77.         }
    
78. 
    
79. done:
    
80.         rcu_read_unlock();
    
81. //如果上面失败，则返回-1.
    
82.         return cpu;
    
83. }

复制代码

然后是enqueue_to_backlog这个方法，首先我们知道在每个cpu都有一个softnet结构，而他有一个input_pkt_queue的队列，以前这个主要是用于非NAPi的驱动的，而这个patch则将这个队列也用与NAPI的处理中了。也就是每个cpu现在都会有一个input_pkt_queue队列，用于保存需要处理的数据包队列。这个队列作用现在是，如果发现不属于当前cpu处理的数据包，则我们可以直接将数据包挂载到他所属的cpu的input_pkt_queue中。

enqueue_to_backlog接受一个skb和cpu为参数，通过cpu来判断skb如何处理。要么加入所属的input_pkt_queue中，要么schecule 软中断。

还有个要注意就是我们知道NAPI为了兼容非NAPI模式，有个backlog的napi_struct结构，也就是非NAPI驱动会schedule backlog这个napi结构，而在enqueue_to_backlog中则是利用了这个结构，也就是它会schedule backlog，因为它会将数据放到input_pkt_queue中，而backlog的pool方法process_backlog就是从input_pkt_queue中取得数据然后交给上层处理。

这里还有一个会用到结构就是 rps_remote_softirq_cpus，它主要是保存了当前cpu上需要去另外的cpu schedule 软中断的cpu 掩码。因为我们可能将要处理的数据包放到了另外的cpu的input queue上，因此我们需要schedule 另外的cpu上的napi(也就是软中断),所以我们需要保存对应的cpu掩码，以便于后面遍历，然后schedule。

而这里为什么mask有两个元素，注释写的很清楚：

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

复制代码

然后就是enqueue_backlog这个函数：

1. static int enqueue_to_backlog(struct sk_buff *skb, int cpu)
   
2. {
   
3.         struct softnet_data *queue;
   
4.         unsigned long flags;
   
5. //取出传递进来的cpu的softnet-data结构
   
6.         queue = &per_cpu(softnet_data, cpu);
   
7. 
   
8.         local_irq_save(flags);
   
9.         __get_cpu_var(netdev_rx_stat).total++;
   
10. //自旋锁
    
11.         spin_lock(&queue->input_pkt_queue.lock);
    
12. //如果保存的队列还没到上限
    
13.         if (queue->input_pkt_queue.qlen <= netdev_max_backlog) {
    
14. //如果当前队列的输入队列长度不为空
    
15.                 if (queue->input_pkt_queue.qlen) {
    
16. enqueue:
    
17. //将数据包加入到input_pkt_queue中,这里会有一个小问题，我们后面再说。
    
18.                         __skb_queue_tail(&queue->input_pkt_queue, skb);
    
19.                         spin_unlock_irqrestore(&queue->input_pkt_queue.lock,
    
20.                             flags);
    
21.                         return NET_RX_SUCCESS;
    
22.                 }
    
23. 
    
24.                 /* Schedule NAPI for backlog device */
    
25. //如果可以调度软中断
    
26.                 if (napi_schedule_prep(&queue->backlog)) {
    
27. //首先判断数据包该不该当前的cpu处理
    
28.                         if (cpu != smp_processor_id()) {
    
29. //如果不该，
    
30.                                 struct rps_remote_softirq_cpus *rcpus =
    
31.                                     &__get_cpu_var(rps_remote_softirq_cpus);
    
32. 
    
33.                                 cpu_set(cpu, rcpus->mask[rcpus->select]);
    
34.                                 __raise_softirq_irqoff(NET_RX_SOFTIRQ);
    
35.                         } else
    
36. //如果就是应该当前cpu处理，则直接schedule 软中断，这里可以看到传递进去的是backlog
    
37.                                 __napi_schedule(&queue->backlog);
    
38.                 }
    
39.                 goto enqueue;
    
40.         }
    
41. 
    
42.         spin_unlock(&queue->input_pkt_queue.lock);
    
43. 
    
44.         __get_cpu_var(netdev_rx_stat).dropped++;
    
45.         local_irq_restore(flags);
    
46. 
    
47.         kfree_skb(skb);
    
48.         return NET_RX_DROP;
    
49. }

复制代码

这里会有一个小问题，那就是假设此时一个属于cpu0的包进入处理，此时我们运行在cpu1,此时将数据包加入到input队列，然后cpu0上面刚好又来了一个cpu0需要处理的数据包，此时由于qlen不为0则又将数据包加入到input队列中，我们会发现cpu0上的napi没机会进行调度了。

google的patch对这个是这样处理的，在软中断处理函数中当数据包处理完毕，会调用net_rps_action来调度前面保存到其他cpu上的input队列。

下面就是代码片断（net_rx_action）

1. //得到对应的rcpus.
   
2. rcpus = &__get_cpu_var(rps_remote_softirq_cpus);
   
3.         select = rcpus->select;
   
4. //翻转select，防止和enqueue_backlog冲突
   
5.         rcpus->select ^= 1;
   
6. 
   
7. //打开中断，此时下面的调度才会起作用.
   
8.         local_irq_enable();
   
9. //这个函数里面调度对应的远程cpu的napi.
   
10.         net_rps_action(&rcpus->mask[select]);

复制代码

然后就是net_rps_action，这个函数很简单，就是遍历所需要处理的cpu，然后调度napi

1. static void net_rps_action(cpumask_t *mask)
   
2. {
   
3.         int cpu;
   
4. 
   
5.         /* Send pending IPI's to kick RPS processing on remote cpus. */
   
6. //遍历
   
7.         for_each_cpu_mask_nr(cpu, *mask) {
   
8.                 struct softnet_data *queue = &per_cpu(softnet_data, cpu);
   
9.                 if (cpu_online(cpu))
   
10. //到对应的cpu调用csd方法。
    
11.                         __smp_call_function_single(cpu, &queue->csd, 0);
    
12.         }
    
13. //清理mask
    
14.         cpus_clear(*mask);
    
15. }

复制代码

上面我们看到会调用csd方法，而上面的csd回掉就是被初始化为trigger_softirq函数。

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

复制代码

上面的函数都分析完毕了，剩下的就很简单了。

首先来看netif_rx如何被修改的，它被修改的很简单，首先是得到当前skb所应该被处理的cpu id，然后再通过比较这个cpu和当前正在处理的cpu id进行比较来做不同的处理。

1. int netif_rx(struct sk_buff *skb)
   
2. {
   
3.         int cpu;
   
4. 
   
5.         /* if netpoll wants it, pretend we never saw it */
   
6.         if (netpoll_rx(skb))
   
7.                 return NET_RX_DROP;
   
8. 
   
9.         if (!skb->tstamp.tv64)
   
10.                 net_timestamp(skb);
    
11. //得到cpu id。
    
12.         cpu = get_rps_cpu(skb->dev, skb);
    
13.         if (cpu < 0)
    
14.                 cpu = smp_processor_id();
    
15. //通过cpu进行队列不同的处理
    
16.         return enqueue_to_backlog(skb, cpu);
    
17. }

复制代码

然后是netif_receive_skb,这里patch将内核本身的这个函数改写为__netif_receive_skb。然后当返回值小于0,则说明不需要对队列进行处理，此时直接发送到3层。

1. int netif_receive_skb(struct sk_buff *skb)
   
2. {
   
3.         int cpu;
   
4. 
   
5.         cpu = get_rps_cpu(skb->dev, skb);
   
6. 
   
7.         if (cpu < 0)
   
8.                 return __netif_receive_skb(skb);
   
9.         else
   
10.                 return enqueue_to_backlog(skb, cpu);
    
11. }

复制代码

最后来总结一下，可以看到input_pkt_queue是一个FIFO的队列，而且如果当qlen有值的时候，也就是在另外的cpu有数据包放到input_pkt_queue中，则当前cpu不会调度napi，而是将数据包放到input_pkt_queue中，然后等待trigger_softirq来调度napi。

因此这个patch完美的解决了软中断在多核下的均衡问题，并且由于是同一个连接会map到相同的cpu，并且input_pkt_queue的使用，因此乱序的问题也不会出现。

我们部门有人测试了这个patch，据说能差不多提高20%。

我看了下35的内核，这个补丁还有另外一个也是这个作者的补丁RFS(rps的增强)都已经进入内核了，看来35的网络性能还是值得期待的。

Godbach

多谢LZ的分享。高亮一把先。

Godbach

标题可以写的更为容易理解的。。。

platinum

看来 35 的网络性能还是值得期待的

platinum

spin_lock(&queue->input_pkt_queue.lock);
之后怎么
spin_unlock_irqrestore(&queue->input_pkt_queue.lock, flags);
可以这么用吗？

simohayha_cu

spin_lock(&queue->input_pkt_queue.lock);
之后怎么
spin_unlock_irqrestore(&queue->input_pkt_queue.l ...
platinum 发表于 2010-07-29 12:59

上面不是还有一个local_irq_save(flags); 呢?

simohayha_cu

标题可以写的更为容易理解的。。。
Godbach 发表于 2010-07-29 10:17

找不到什么好的题目。。

Godbach

这个patch是为了实现中断均衡的功能吧。

platinum

上面不是还有一个local_irq_save(flags); 呢?
simohayha_cu 发表于 2010-07-29 13:49

哦，是我没说清楚
我的疑问在于，对于通一个 lock 变量，是否可以采用不同的 lock 方式
比如在有些代码中，看到了 spin_lock(lock)，又同时看到有 spin_lock_bh(lock)

0vk0

多谢LZ的分享，学习了