Postgresql实验系列（4）SIMD提升线性搜索性能24.5%（附带PG SIMD完整用例）

概要

接上一篇《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();
1
2
3
4

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

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

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

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

性能差距（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
1
2
3
4
5

结果：

• 场景一：（使用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;
}
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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

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);
#else
	memcpy(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);
#else
	return ~UINT64CONST(0) / 0xFF * c;
#endif
}

#ifndef USE_NO_SIMD
static inline Vector32
vector32_broadcast(const uint32 c)
{
#ifdef USE_SSE2
	return _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_CHECKING
	bool		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));
#else
	result = vector8_is_highbit_set(vector8_eq(v, vector8_broadcast(c)));
#endif

	return 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);
#else
	return 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_CHECKING
	bool		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)));
#endif

	return result;
}

/*
 * Return true if the high bit of any element is set
 */
static inline bool
vector8_is_highbit_set(const Vector8 v)
{
#ifdef USE_SSE2
	return _mm_movemask_epi8(v) != 0;
#elif defined(USE_NEON)
	return vmaxvq_u8(v) > 0x7F;
#else
	return 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);
#else
	return 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_SSE2
	return _mm_or_si128(v1, v2);
#elif defined(USE_NEON)
	return vorrq_u8(v1, v2);
#else
	return v1 | v2;
#endif
}

#ifndef USE_NO_SIMD
static inline Vector32
vector32_or(const Vector32 v1, const Vector32 v2)
{
#ifdef USE_SSE2
	return _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_SSE2
	return _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_SSE2
	return _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_SSE2
	return _mm_cmpeq_epi32(v1, v2);
#elif defined(USE_NEON)
	return vceqq_u32(v1, v2);
#endif
}
#endif							/* ! USE_NO_SIMD */

#endif							/* SIMD_H */

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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374