Linux XPS实现

概念

  • XPS(Transmit Packet Steering)是什么?
    XPS是RPS在发送路径的实现。
    通过配置 /sys/class/net//queues/tx-/xps_cpus
    设置了能够使用某个发送队列的cpu集合。

  • Why XPS?
    如果过多的cpu使用相同的tx队列,那么加重tx对应的qdisc锁的争抢。 如果每个cpu只关联了一个tx,甚至能消除竞争
    可以减小因为发送完成中断造成的cache miss。
    因此xps_cpus的配置最好结合/proc/irq//smp_affinity, 映射最好在同一个cpu或者同一个numa node的cpu上

netdev_pick_tx

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
//在__dev_queue_xmit中选择发送队列的时候被调用
struct netdev_queue *netdev_pick_tx(struct net_device *dev,
struct sk_buff *skb,
void *accel_priv)
{
int queue_index = 0;
#ifdef CONFIG_XPS
u32 sender_cpu = skb->sender_cpu - 1;
if (sender_cpu >= (u32)NR_CPUS)
skb->sender_cpu = raw_smp_processor_id() + 1; //设置sender_cpu为当前cpu号+1
#endif
if (dev->real_num_tx_queues != 1) { //多队列调用驱动的ndo_select_queue函数
const struct net_device_ops *ops = dev->netdev_ops;
if (ops->ndo_select_queue)
queue_index = ops->ndo_select_queue(dev, skb, accel_priv,
__netdev_pick_tx); //实际都会调用__netdev_pick_tx进行处理
else
queue_index = __netdev_pick_tx(dev, skb); //使用xps或者默认算法,选择发送队列
if (!accel_priv)
queue_index = netdev_cap_txqueue(dev, queue_index);
}
skb_set_queue_mapping(skb, queue_index); //skb->queue_mapping = queue_index
return netdev_get_tx_queue(dev, queue_index); //返回选择的这个tx
}
static u16 __netdev_pick_tx(struct net_device *dev, struct sk_buff *skb)
{
struct sock *sk = skb->sk;
int queue_index = sk_tx_queue_get(sk); //获取上一次的发送队列
if (queue_index < 0 || skb->ooo_okay || //未选择发送队列; 或者qdisc没有该sk的数据包,则可以重新选择发送队列
queue_index >= dev->real_num_tx_queues) { // queue_index值非法
int new_index = get_xps_queue(dev, skb); //通过xps算法获取发送队列
if (new_index < 0)
new_index = skb_tx_hash(dev, skb); //使用默认算法,优先使用rx对应的tx队列,否则使用hash的方式映射
if (queue_index != new_index && sk &&
sk_fullsock(sk) &&
rcu_access_pointer(sk->sk_dst_cache))
sk_tx_queue_set(sk, new_index); //更新sk_tx_queue_mapping
queue_index = new_index;
}
return queue_index; //返回选择发送队列
}

get_xps_queue

在get_xps_queue中进行xps算法,选择发送队列
如果当前cpu关联到多个发送队列,则通过hash来选择一个

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
static inline int get_xps_queue(struct net_device *dev, struct sk_buff *skb)
{
#ifdef CONFIG_XPS
struct xps_dev_maps *dev_maps;
struct xps_map *map;
int queue_index = -1;
rcu_read_lock();
dev_maps = rcu_dereference(dev->xps_maps); //配置了xps映射
if (dev_maps) {
map = rcu_dereference(
dev_maps->cpu_map[skb->sender_cpu - 1]);
if (map) {
if (map->len == 1)
queue_index = map->queues[0]; //这个cpu只关联到一个发送队列,直接选择
else
queue_index = map->queues[reciprocal_scale(skb_get_hash(skb),
map->len)]; //这个cpu关联到多个发送队列,通过hash选择
if (unlikely(queue_index >= dev->real_num_tx_queues))
queue_index = -1;
}
}
rcu_read_unlock();
return queue_index;
#else
return -1;
#endif
}

xps_dev_maps

/sys/class/net//queues/tx-/xps_cpus配置了tx队列到cpu集合的映射
实际上代码内使用了反向映射,通过xps_dev_maps存放到cpu到tx队列集合的映射

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
struct xps_map {
unsigned int len;
unsigned int alloc_len;
struct rcu_head rcu;
u16 queues[0]; //tx队列的集合
};
#define XPS_MAP_SIZE(_num) (sizeof(struct xps_map) + ((_num) * sizeof(u16)))
#define XPS_MIN_MAP_ALLOC ((L1_CACHE_ALIGN(offsetof(struct xps_map, queues[1])) \
- sizeof(struct xps_map)) / sizeof(u16))
/*
* This structure holds all XPS maps for device. Maps are indexed by CPU.
*/
struct xps_dev_maps {
struct rcu_head rcu;
struct xps_map __rcu *cpu_map[0]; //索引为cpu
};

参考资料

xps: Transmit Packet Steering
Documentation/networking/scaling.txt