vppinfra--字节序转换、bitops、cacheline、jmp机制

dpdk-vpp源码解读

发布于 2023-03-07 17:03:26

7510

发布于 2023-03-07 17:03:26

文章被收录于专栏：DPDK VPP源码分析

本篇介绍一些vpp代码中经常使用一些宏定义，了解一下内部的一些实现。

字节序转换相关api

vpp提供的字节序相关的api都定义在src\vppinfra\byte_order.h文件中，支持u16、i16、u32、i32、u64、i64字节序的转换。下面是定义当前系统是大端还是小端的宏定义

/*系统大小端设置宏*/
#if (__BYTE_ORDER__)==( __ORDER_LITTLE_ENDIAN__)
#define CLIB_ARCH_IS_BIG_ENDIAN (0)
#define CLIB_ARCH_IS_LITTLE_ENDIAN (1)
#else
/* Default is big endian. */
#define CLIB_ARCH_IS_BIG_ENDIAN (1)
#define CLIB_ARCH_IS_LITTLE_ENDIAN (0)
#endif

宏定义BYTE_ORDER__和__ORDER_LITTLE_ENDIAN在vpp工程中是搜索不到的。其实这些宏是gcc默认的宏定义选项。我们可以在自己的系统上使用下面的命令来查询（当前环境是小端模式）

$: gcc -E -dD -xc /dev/null | grep ENDIAN
#define __ORDER_LITTLE_ENDIAN__ 1234
#define __ORDER_BIG_ENDIAN__ 4321
#define __ORDER_PDP_ENDIAN__ 3412
/*这是我们需要的宏定义，当前环境设置为小端模式*/
#define __BYTE_ORDER__ __ORDER_LITTLE_ENDIAN__
#define __FLOAT_WORD_ORDER__ __ORDER_LITTLE_ENDIAN__

我们以u16主机序到网络序的转换函数：clib_net_to_host_u16来举例，来看下宏定义展开后的相应函数实现如下：

#define clib_arch_is_big_endian    CLIB_ARCH_IS_BIG_ENDIAN
 always_inline u16
clib_byte_swap_u16 (u16 x)
{
    /*使用gcc的buildin字节转换函数*/
    return __builtin_bswap16 (x);
}

 always_inline u16
clib_host_to_big_u16 (u16 x)
{
    /*如果当前系统不是大端才进行字节转换*/
    if (! clib_arch_is_big_endian)
       x = clib_byte_swap_u16 (x);
    return x;
}

 always_inline u16    
clib_big_to_host_u16 (u16 x)
{ 
    return clib_host_to_big_u16 (x); 
}    
23   /*默认网络序是大端模式，所以这里调用big相应的函数*/
always_inline u16
clib_net_to_host_u16 (u16 x)
{ 
    return clib_big_to_host_u16 (x); 
}

最终调用的gcc的内置（buildin）函数

/*按照字节进行反转，返回反转后的结果*/
uint16_t __builtin_bswap16 (uint16_t x)

bitops相关api

bit操作相关函数在src\vppinfra\bitops.h文件中，有些函数还是满高效的。

1、popcnt是“population count”的缩写，该操作一般翻译为“位1计数”，即统计有多少个“为1的位”。在编译器没有内置支持的时候，使用了Hacker's Delight算法。

always_inline uword
count_set_bits (uword x)
{
#ifdef __POPCNT__ /*gcc默认宏定义吧，目前再我自己环境上没有设置*/
#if uword_bits == 64
  return __builtin_popcountll (x);
#else
  return __builtin_popcount (x);
#endif
#else
#if uword_bits == 64
  const uword c1 = 0x5555555555555555;
  const uword c2 = 0x3333333333333333;
  const uword c3 = 0x0f0f0f0f0f0f0f0f;
#else
  const uword c1 = 0x55555555;
  const uword c2 = 0x33333333;
  const uword c3 = 0x0f0f0f0f;
#endif

  /* Sum 1 bit at a time. */
  x = x - ((x >> (uword) 1) & c1);

  /* 2 bits at a time. */
  x = (x & c2) + ((x >> (uword) 2) & c2);

  /* 4 bits at a time. */
  x = (x + (x >> (uword) 4)) & c3;

  /* 8, 16, 32 bits at a time. */
  x = x + (x >> (uword) 8);
  x = x + (x >> (uword) 16);
#if uword_bits == 64
  x = x + (x >> (uword) 32);
#endif

  return x & (2 * BITS (uword) - 1);
#endif
}

2、变量指定bit进行翻转。目前搜索了一下使用的地方在ipv6_get_key 函数计算hash数值中有使用：跟踪了一下函数调用，在handoff功能中有使用，是一个软件实现rss算法。根据报文三元组（src、dip、ptotocol）。

always_inline uword
rotate_left (uword x, uword i)
{
  return (x << i) | (x >> (BITS (i) - i));
}
always_inline uword
rotate_right (uword x, uword i)
{
  return (x >> i) | (x << (BITS (i) - i));
}
/*根据ipv6报文头计算hash数值*/
static inline u64
ipv6_get_key (ip6_header_t * ip)
{
  u64 hash_key;

  hash_key = ip->src_address.as_u64[0] ^
    rotate_left (ip->src_address.as_u64[1], 13) ^
    rotate_left (ip->dst_address.as_u64[0], 26) ^
    rotate_left (ip->dst_address.as_u64[1], 39) ^ ip->protocol;

  return hash_key;
}

3、遍历整数中对应的bit位，进行相应的操作。

/* var : 获取相应bit位对应的数值大小
 * mask :对应需要处理的bit位的掩码
 * body： 处理算法
 **/
#define foreach_set_bit(var,mask,body)                    \
do {                                    \
  uword _foreach_set_bit_m_##var = (mask);                \
  uword _foreach_set_bit_f_##var;                    \
  while (_foreach_set_bit_m_##var != 0)                    \
    {                                    \
      _foreach_set_bit_f_##var = first_set (_foreach_set_bit_m_##var);    \
      _foreach_set_bit_m_##var ^= _foreach_set_bit_f_##var;        \
      (var) = min_log2 (_foreach_set_bit_f_##var);            \
      do { body; } while (0);                        \
    }                                    \
} while (0)

  /* *INDENT-OFF feature使能去使能中使用，根据bit位做相应的设置* */
  foreach_set_bit (feat, diff_fb,
  ({
    vnet_feature_enable_disable (gbp_itf_feat_bit_pos_to_arc[feat],
                                 gbp_itf_feat_bit_pos_to_feat[feat],
                                 gi->gi_sw_if_index, 0, 0, 0);
  }));

cache相关

在编译器没有指定cache line大小的情况下，我们将它指定成64字节。不允许编译器指定的cache line超过256字节。 cache line bytes指的是一次性从内存读入到CPU缓存中的字节数目。CPU访问自己内部的缓存比访问内存的效率高得多。 cache line对齐的目的是避免cache一致性的问题。

/*查询当前系统cacheline大小*/
[19:53:34]root:src$ cat /sys/devices/system/cpu/cpu1/cache/index1/coherency_line_size
64

cacheline相应的宏定义：在文件src\vppinfra\cache.h中。

/*
 * Allow CFLAGS to override the configured / deduced cache line size
 */
#ifndef CLIB_LOG2_CACHE_LINE_BYTES

/* Default cache line size of 64 bytes. */
#ifndef CLIB_LOG2_CACHE_LINE_BYTES
#define CLIB_LOG2_CACHE_LINE_BYTES 6
#endif

#endif /* CLIB_LOG2_CACHE_LINE_BYTES defined */

#if (CLIB_LOG2_CACHE_LINE_BYTES >= 9)
#error Cache line size 512 bytes or greater
#endif
/*定义cacheline大小宏定义，默认是64字节*/
#define CLIB_CACHE_LINE_BYTES (1 << CLIB_LOG2_CACHE_LINE_BYTES)

在很多结构体中都包含的这个宏CLIB_CACHE_LINE_ALIGN_MARK，具体的意义是什么呢？目的是确保从此处开始的内存地址必须是 cacheline 的整数倍。

#define CLIB_CACHE_LINE_ALIGN_MARK(mark) u8 mark[0] __attribute__((aligned(CLIB_CACHE_LINE_BYTES)))

CLIB_CACHE_LINE_ROUND宏需要将X设置成cacheline大小的倍数。

#define CLIB_CACHE_LINE_ROUND(x) ((x + CLIB_CACHE_LINE_BYTES - 1) & ~(CLIB_CACHE_LINE_BYTES - 1))

下面是cache预取相关的函数。 Cache之所以能够提高系统性能，主要是程序执行存在局部性现象，即时间局部性和空间局部性。 1）时间局部性：是指程序即将用到的指令/数据可能就是目前正在使用的指令/数据。因此，当前用到的指令/数据在使用完毕之后可以暂时存放在Cache中，可以在将来的时候再被处理器用到。一个简单的例子就是一个循环语句的指令，当循环终止的条件满足之前，处理器需要反复执行循环语句中的指令。 2）空间局部性：是指程序即将用到的指令/数据可能与目前正在使用的指令/数据在空间上相邻或者相近。因此，在处理器处理当前指令/数据时，可以从内存中把相邻区域的指令/数据读取到Cache中，这样，当处理器需要处理相邻内存区域的指令/数据时，可以直接从Cache中读取，节省访问内存的时间。一个简单的例子就是一个需要顺序处理的数组。

/* Read/write arguments to __builtin_prefetch. */
#define CLIB_PREFETCH_READ 0
#define CLIB_PREFETCH_LOAD 0    /* alias for read */
#define CLIB_PREFETCH_WRITE 1
#define CLIB_PREFETCH_STORE 1    /* alias for write */

#define _CLIB_PREFETCH(n,size,type)                \
  if ((size) > (n)*CLIB_CACHE_LINE_BYTES)            \
    __builtin_prefetch (_addr + (n)*CLIB_CACHE_LINE_BYTES,    \
            CLIB_PREFETCH_##type,           \
            /* locality */ 3);

#define CLIB_PREFETCH(addr,size,type)        \
do {                        \
  void * _addr = (addr);            \
                        \
  ASSERT ((size) <= 4*CLIB_CACHE_LINE_BYTES);    \
  _CLIB_PREFETCH (0, size, type);        \
  _CLIB_PREFETCH (1, size, type);        \
  _CLIB_PREFETCH (2, size, type);        \
  _CLIB_PREFETCH (3, size, type);        \
} while (0)

#undef _
/**/
static_always_inline void
clib_prefetch_load (void *p)
{
  CLIB_PREFETCH (p, CLIB_CACHE_LINE_BYTES, LOAD);
}
/**/
static_always_inline void
clib_prefetch_store (void *p)
{
  CLIB_PREFETCH (p, CLIB_CACHE_LINE_BYTES, STORE);
}

协程相关api-longjmp

vpp process类型的node使用jmp机制相关的api实现协程的功能。对应文件src\vppinfra\longjmp.[hs]。相关函数的定义是使用汇编语言写的。每个process类型node是由jump机制构成的一个协程，协程主要用于等待、处理事件。使用longjmp/setjmp的轻量级多任务协程，由应用进程自行进行调度，不受操作系统调度机制的影响，上下文切换只损耗调用longjmp/setjmp的时间。协程中运行的函数类似于线程函数，区别在于协程函数Can be suspended, wait for events, be resumed… (based on setjump/longjump).

/* 跳转到setjmp设置点，reurn_value 就是跳转到clib—setjmp返回数值，*/
void clib_longjmp (clib_longjmp_t * save, uword return_value);

/* Save context.  Returns given value if jump is not taken;
   otherwise returns value from clib_longjmp if long jump is taken. 
   设置跳转返回点，默认返回*/
uword clib_setjmp (clib_longjmp_t * save, uword return_value_not_taken);

/* 在设置的stack区中跳转，运行fuc函数，函数返回值就是func的返回值*/
uword clib_calljmp (uword (*func) (uword func_arg),
            uword func_arg, void *stack);

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-11-02，如有侵权请联系 cloudcommunity@tencent.com 删除

编程算法