Postgresql实验系列(4)SIMD提升线性搜索性能24.5%(附带PG SIMD完整用例)
创始人
2024-01-25 02:06:16
0

概要

接上一篇《Postgresql引入SIMD指令集》

PG引入SIMD执行集后具体有多大性能提升?本篇抽取PG的simd库,对比线性搜索场景的性能:

测试场景(文章最后提供完整程序)

构造一个存有14亿数字的数组

	uint32 cnt = 1410065408;uint32 *xids = (uint32 *) malloc(sizeof(uint32) * cnt);for (int i = 0; i < cnt; i++)xids[i] = rand();

场景一:使用SIMD查询一个数字是否在数组中,重复10次

// do 10 times!for (int i = 0; i < 10; i++)res = pg_lfind32(xid, xids, cnt);

场景二:直接遍历查询一个数字是否在数组中,重复10次

	// do 10 times!for (int i = 0; i < 10; i++)for (int i = 0; i < cnt; i++)if (xid == xids[i])res = true;

性能差距(on AMD EPYC 7K62 48-C Processor)

执行14亿数组查询,执行结果:

在这里插入图片描述

[mingjie@VM-130-23-centos ~/proj/simd][PGROOT99:9901]$ ./conftest sample: 1804289383 846930886 1681692777...
[use SIMD]find x among 1410065408 numbers: 27.480000 seconds
[use no SIMD]find x among 1410065408 numbers:  34.220000 seconds

结果:

  • 场景一:(使用SIMD从14亿数字中查询一个数是否存在) * 10次
    • 时间 = 27.480000 秒
  • 场景二:(直接遍历14亿数字查询一个数是否存在) * 10次
    • 时间 = 34.220000 秒

性能差距:SIMD在该场景有 24.5%的性能提升。


测试程序 & 编译方法

编译:

gcc -o conftest -Wall -g -ggdb -O0 -g3 -gdwarf-2 -msse4.2 main.c

main.c

/***************************************************/
#include 
#include 
#include 
#include 
#include /***************************************************/
#define Assert(condition) assert(condition)
#define true	((bool) 1)
#define false	((bool) 0)
typedef unsigned char uint8;	/* == 8 bits */
typedef unsigned short uint16;	/* == 16 bits */
typedef unsigned int uint32;	/* == 32 bits */
typedef unsigned char bool;
typedef size_t Size;
/***************************************************/
// #define USE_ASSERT_CHECKING#include "simd.h"
/***************************************************//** pg_lfind32** Return true if there is an element in 'base' that equals 'key', otherwise* return false.*/
static inline bool
pg_lfind32(uint32 key, uint32 *base, uint32 nelem)
{uint32		i = 0;#ifndef USE_NO_SIMD/** For better instruction-level parallelism, each loop iteration operates* on a block of four registers.  Testing for SSE2 has showed this is ~40%* faster than using a block of two registers.*/const Vector32 keys = vector32_broadcast(key);	/* load copies of key */const uint32 nelem_per_vector = sizeof(Vector32) / sizeof(uint32);const uint32 nelem_per_iteration = 4 * nelem_per_vector;/* round down to multiple of elements per iteration */const uint32 tail_idx = nelem & ~(nelem_per_iteration - 1);for (i = 0; i < tail_idx; i += nelem_per_iteration){Vector32	vals1,vals2,vals3,vals4,result1,result2,result3,result4,tmp1,tmp2,result;/* load the next block into 4 registers */vector32_load(&vals1, &base[i]);vector32_load(&vals2, &base[i + nelem_per_vector]);vector32_load(&vals3, &base[i + nelem_per_vector * 2]);vector32_load(&vals4, &base[i + nelem_per_vector * 3]);/* compare each value to the key */result1 = vector32_eq(keys, vals1);result2 = vector32_eq(keys, vals2);result3 = vector32_eq(keys, vals3);result4 = vector32_eq(keys, vals4);/* combine the results into a single variable */tmp1 = vector32_or(result1, result2);tmp2 = vector32_or(result3, result4);result = vector32_or(tmp1, tmp2);/* see if there was a match */if (vector32_is_highbit_set(result)){return true;}}
#endif							/* ! USE_NO_SIMD *//* Process the remaining elements one at a time. */for (; i < nelem; i++){if (key == base[i])return true;}return false;
}
/***************************************************/int main()
{clock_t start, finish;double  duration;uint32 xid = 11;uint32 cnt = 1410065408;uint32 *xids = (uint32 *) malloc(sizeof(uint32) * cnt);bool res = false;for (int i = 0; i < cnt; i++)xids[i] = rand();printf("sample: %d %d %d...\n", xids[0], xids[1], xids[2]);/***************************************************//* SIMD *//***************************************************/start = clock();// do 10 times!for (int i = 0; i < 10; i++)res = pg_lfind32(xid, xids, cnt);finish = clock();  duration = (double)(finish - start) / CLOCKS_PER_SEC;printf( "[use SIMD]find x among %d numbers: %f seconds\n", cnt, duration );// printf("res: %d\n", res);/***************************************************//* no SIMD *//***************************************************/start = clock();// do 10 times!for (int i = 0; i < 10; i++)for (int i = 0; i < cnt; i++)if (xid == xids[i])res = true;finish = clock();  duration = (double)(finish - start) / CLOCKS_PER_SEC;printf( "[use no SIMD]find x among %d numbers:  %f seconds\n", cnt, duration );// printf("res: %d\n", res);return res;
}

simd.h(from postgresql15)

/*-------------------------------------------------------------------------** simd.h*	  Support for platform-specific vector operations.** Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group* Portions Copyright (c) 1994, Regents of the University of California** src/include/port/simd.h** NOTES* - VectorN in this file refers to a register where the element operands* are N bits wide. The vector width is platform-specific, so users that care* about that will need to inspect "sizeof(VectorN)".**-------------------------------------------------------------------------*/
#ifndef SIMD_H
#define SIMD_H#if (defined(__x86_64__) || defined(_M_AMD64))
/** SSE2 instructions are part of the spec for the 64-bit x86 ISA. We assume* that compilers targeting this architecture understand SSE2 intrinsics.** We use emmintrin.h rather than the comprehensive header immintrin.h in* order to exclude extensions beyond SSE2. This is because MSVC, at least,* will allow the use of intrinsics that haven't been enabled at compile* time.*/
#include 
#define USE_SSE2
typedef __m128i Vector8;
typedef __m128i Vector32;#elif defined(__aarch64__) && defined(__ARM_NEON)
/** We use the Neon instructions if the compiler provides access to them (as* indicated by __ARM_NEON) and we are on aarch64.  While Neon support is* technically optional for aarch64, it appears that all available 64-bit* hardware does have it.  Neon exists in some 32-bit hardware too, but we* could not realistically use it there without a run-time check, which seems* not worth the trouble for now.*/
#include 
#define USE_NEON
typedef uint8x16_t Vector8;
typedef uint32x4_t Vector32;#else
/** If no SIMD instructions are available, we can in some cases emulate vector* operations using bitwise operations on unsigned integers.  Note that many* of the functions in this file presently do not have non-SIMD* implementations.  In particular, none of the functions involving Vector32* are implemented without SIMD since it's likely not worthwhile to represent* two 32-bit integers using a uint64.*/
#define USE_NO_SIMD
typedef uint64 Vector8;
#endif/* load/store operations */
static inline void vector8_load(Vector8 *v, const uint8 *s);
#ifndef USE_NO_SIMD
static inline void vector32_load(Vector32 *v, const uint32 *s);
#endif/* assignment operations */
static inline Vector8 vector8_broadcast(const uint8 c);
#ifndef USE_NO_SIMD
static inline Vector32 vector32_broadcast(const uint32 c);
#endif/* element-wise comparisons to a scalar */
static inline bool vector8_has(const Vector8 v, const uint8 c);
static inline bool vector8_has_zero(const Vector8 v);
static inline bool vector8_has_le(const Vector8 v, const uint8 c);
static inline bool vector8_is_highbit_set(const Vector8 v);
#ifndef USE_NO_SIMD
static inline bool vector32_is_highbit_set(const Vector32 v);
#endif/* arithmetic operations */
static inline Vector8 vector8_or(const Vector8 v1, const Vector8 v2);
#ifndef USE_NO_SIMD
static inline Vector32 vector32_or(const Vector32 v1, const Vector32 v2);
static inline Vector8 vector8_ssub(const Vector8 v1, const Vector8 v2);
#endif/** comparisons between vectors** Note: These return a vector rather than boolean, which is why we don't* have non-SIMD implementations.*/
#ifndef USE_NO_SIMD
static inline Vector8 vector8_eq(const Vector8 v1, const Vector8 v2);
static inline Vector32 vector32_eq(const Vector32 v1, const Vector32 v2);
#endif/** Load a chunk of memory into the given vector.*/
static inline void
vector8_load(Vector8 *v, const uint8 *s)
{
#if defined(USE_SSE2)*v = _mm_loadu_si128((const __m128i *) s);
#elif defined(USE_NEON)*v = vld1q_u8(s);
#elsememcpy(v, s, sizeof(Vector8));
#endif
}#ifndef USE_NO_SIMD
static inline void
vector32_load(Vector32 *v, const uint32 *s)
{
#ifdef USE_SSE2*v = _mm_loadu_si128((const __m128i *) s);
#elif defined(USE_NEON)*v = vld1q_u32(s);
#endif
}
#endif							/* ! USE_NO_SIMD *//** Create a vector with all elements set to the same value.*/
static inline Vector8
vector8_broadcast(const uint8 c)
{
#if defined(USE_SSE2)return _mm_set1_epi8(c);
#elif defined(USE_NEON)return vdupq_n_u8(c);
#elsereturn ~UINT64CONST(0) / 0xFF * c;
#endif
}#ifndef USE_NO_SIMD
static inline Vector32
vector32_broadcast(const uint32 c)
{
#ifdef USE_SSE2return _mm_set1_epi32(c);
#elif defined(USE_NEON)return vdupq_n_u32(c);
#endif
}
#endif							/* ! USE_NO_SIMD *//** Return true if any elements in the vector are equal to the given scalar.*/
static inline bool
vector8_has(const Vector8 v, const uint8 c)
{bool		result;/* pre-compute the result for assert checking */
#ifdef USE_ASSERT_CHECKINGbool		assert_result = false;for (Size i = 0; i < sizeof(Vector8); i++){if (((const uint8 *) &v)[i] == c){assert_result = true;break;}}
#endif							/* USE_ASSERT_CHECKING */#if defined(USE_NO_SIMD)/* any bytes in v equal to c will evaluate to zero via XOR */result = vector8_has_zero(v ^ vector8_broadcast(c));
#elseresult = vector8_is_highbit_set(vector8_eq(v, vector8_broadcast(c)));
#endifreturn result;
}/** Convenience function equivalent to vector8_has(v, 0)*/
static inline bool
vector8_has_zero(const Vector8 v)
{
#if defined(USE_NO_SIMD)/** We cannot call vector8_has() here, because that would lead to a* circular definition.*/return vector8_has_le(v, 0);
#elsereturn vector8_has(v, 0);
#endif
}/** Return true if any elements in the vector are less than or equal to the* given scalar.*/
static inline bool
vector8_has_le(const Vector8 v, const uint8 c)
{bool		result = false;/* pre-compute the result for assert checking */
#ifdef USE_ASSERT_CHECKINGbool		assert_result = false;for (Size i = 0; i < sizeof(Vector8); i++){if (((const uint8 *) &v)[i] <= c){assert_result = true;break;}}
#endif							/* USE_ASSERT_CHECKING */#if defined(USE_NO_SIMD)/** To find bytes <= c, we can use bitwise operations to find bytes < c+1,* but it only works if c+1 <= 128 and if the highest bit in v is not set.* Adapted from* https://graphics.stanford.edu/~seander/bithacks.html#HasLessInWord*/if ((int64) v >= 0 && c < 0x80)result = (v - vector8_broadcast(c + 1)) & ~v & vector8_broadcast(0x80);else{/* one byte at a time */for (Size i = 0; i < sizeof(Vector8); i++){if (((const uint8 *) &v)[i] <= c){result = true;break;}}}
#else/** Use saturating subtraction to find bytes <= c, which will present as* NUL bytes.  This approach is a workaround for the lack of unsigned* comparison instructions on some architectures.*/result = vector8_has_zero(vector8_ssub(v, vector8_broadcast(c)));
#endifreturn result;
}/** Return true if the high bit of any element is set*/
static inline bool
vector8_is_highbit_set(const Vector8 v)
{
#ifdef USE_SSE2return _mm_movemask_epi8(v) != 0;
#elif defined(USE_NEON)return vmaxvq_u8(v) > 0x7F;
#elsereturn v & vector8_broadcast(0x80);
#endif
}/** Exactly like vector8_is_highbit_set except for the input type, so it* looks at each byte separately.** XXX x86 uses the same underlying type for 8-bit, 16-bit, and 32-bit* integer elements, but Arm does not, hence the need for a separate* function. We could instead adopt the behavior of Arm's vmaxvq_u32(), i.e.* check each 32-bit element, but that would require an additional mask* operation on x86.*/
#ifndef USE_NO_SIMD
static inline bool
vector32_is_highbit_set(const Vector32 v)
{
#if defined(USE_NEON)return vector8_is_highbit_set((Vector8) v);
#elsereturn vector8_is_highbit_set(v);
#endif
}
#endif							/* ! USE_NO_SIMD *//** Return the bitwise OR of the inputs*/
static inline Vector8
vector8_or(const Vector8 v1, const Vector8 v2)
{
#ifdef USE_SSE2return _mm_or_si128(v1, v2);
#elif defined(USE_NEON)return vorrq_u8(v1, v2);
#elsereturn v1 | v2;
#endif
}#ifndef USE_NO_SIMD
static inline Vector32
vector32_or(const Vector32 v1, const Vector32 v2)
{
#ifdef USE_SSE2return _mm_or_si128(v1, v2);
#elif defined(USE_NEON)return vorrq_u32(v1, v2);
#endif
}
#endif							/* ! USE_NO_SIMD *//** Return the result of subtracting the respective elements of the input* vectors using saturation (i.e., if the operation would yield a value less* than zero, zero is returned instead).  For more information on saturation* arithmetic, see https://en.wikipedia.org/wiki/Saturation_arithmetic*/
#ifndef USE_NO_SIMD
static inline Vector8
vector8_ssub(const Vector8 v1, const Vector8 v2)
{
#ifdef USE_SSE2return _mm_subs_epu8(v1, v2);
#elif defined(USE_NEON)return vqsubq_u8(v1, v2);
#endif
}
#endif							/* ! USE_NO_SIMD *//** Return a vector with all bits set in each lane where the the corresponding* lanes in the inputs are equal.*/
#ifndef USE_NO_SIMD
static inline Vector8
vector8_eq(const Vector8 v1, const Vector8 v2)
{
#ifdef USE_SSE2return _mm_cmpeq_epi8(v1, v2);
#elif defined(USE_NEON)return vceqq_u8(v1, v2);
#endif
}
#endif							/* ! USE_NO_SIMD */#ifndef USE_NO_SIMD
static inline Vector32
vector32_eq(const Vector32 v1, const Vector32 v2)
{
#ifdef USE_SSE2return _mm_cmpeq_epi32(v1, v2);
#elif defined(USE_NEON)return vceqq_u32(v1, v2);
#endif
}
#endif							/* ! USE_NO_SIMD */#endif							/* SIMD_H */

相关内容

热门资讯

新华社快讯:全国基本医保参保率...   新华社快讯:国家医疗保障局局长章轲7月24日在国新办举行的“高质量完成‘十四五’规划”系列主题新...
国际锐评丨封关在即,海南自贸港...   7月23日,中国海南自贸港建设取得重大进展——中国政府宣布,海南自贸港定于今年12月18日正式启...
天气接“地气” 气象科技助力苹...   “咬一口咱延安的苹果,那股子甜劲儿,绝了!”提到延安的苹果,陕西省延安市气象局局长王维刚满脸自豪...
人民财评:夏日“夜经济”燃动消...   晚风起,夜未央。情侣挽手在霓虹灯下逛街,好友在大排档开怀畅饮,消夏露天音乐会释放青春热情,旅游景...
数字经济创新发展高级研修班(国...   由世界互联网大会数字研修院与中国国家互联网信息办公室数据与技术保障中心联合主办的数字经济创新发展...
品尝美食、欣赏歌舞、号脉开方…...   石榴云/新疆日报记者  暮色四合,华灯初上,天山南北的夜市悄然“醒来”,各种摊位亮灯营业。美食、...
各地积极应对强降雨 抢险救援加...   央视新闻客户端消息(新闻联播):受持续强降雨天气影响,福建、湖北境内多条国省干线因山体滑坡导致道...
法润三晋 e法同行|晋秀山河铺...   一部厚重的法治短剧《法度三晋》串起三晋大地几千年悠悠历史,一曲悠扬的现代合唱《晋秀山河 e法同行...
中金黄金发布情况说明 中金黄金7月24日发布情况说明公告: 2025年7月23日,东北大学6名学生在中金黄金股份有限公司(...
杭州通报自来水异味调查情况:依...   新华社杭州7月23日电(记者 马剑、李平)针对杭州市余杭区部分小区发生供水异常的情况,杭州市联合...