xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #28577 from taosdata/enh/TD-32652-main 

enh(query)[TD-32652]: enable AVX implementation with CPU dispatching
This commit is contained in:
Shengliang Guan 2024-11-01 17:30:52 +08:00 committed by GitHub
parent 526904fdd1 647066c4f9
commit 65f41cad20
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 434 additions and 374 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
cmake/cmake.define
View File

@ -177,48 +177,11 @@ ELSE()
SET(COMPILER_SUPPORT_AVX512VL false)
ELSE()
CHECK_C_COMPILER_FLAG("-mfma" COMPILER_SUPPORT_FMA)
CHECK_C_COMPILER_FLAG("-mavx" COMPILER_SUPPORT_AVX)
CHECK_C_COMPILER_FLAG("-mavx2" COMPILER_SUPPORT_AVX2)
CHECK_C_COMPILER_FLAG("-mavx512f" COMPILER_SUPPORT_AVX512F)
CHECK_C_COMPILER_FLAG("-mavx512vbmi" COMPILER_SUPPORT_AVX512BMI)
CHECK_C_COMPILER_FLAG("-mavx512vl" COMPILER_SUPPORT_AVX512VL)
INCLUDE(CheckCSourceRuns)
SET(CMAKE_REQUIRED_FLAGS "-mavx")
check_c_source_runs("
#include <immintrin.h>
int main() {
__m256d a, b, c;
double buf[4] = {0};
a = _mm256_loadu_pd(buf);
b = _mm256_loadu_pd(buf);
c = _mm256_add_pd(a, b);
_mm256_storeu_pd(buf, c);
for (int i = 0; i < sizeof(buf) / sizeof(buf[0]); ++i) {
IF (buf[i] != 0) {
return 1;
}
}
return 0;
}
" COMPILER_SUPPORT_AVX)
SET(CMAKE_REQUIRED_FLAGS "-mavx2")
check_c_source_runs("
#include <immintrin.h>
int main() {
__m256i a, b, c;
int buf[8] = {0};
a = _mm256_loadu_si256((__m256i *)buf);
b = _mm256_loadu_si256((__m256i *)buf);
c = _mm256_and_si256(a, b);
_mm256_storeu_si256((__m256i *)buf, c);
for (int i = 0; i < sizeof(buf) / sizeof(buf[0]); ++i) {
IF (buf[i] != 0) {
return 1;
}
}
return 0;
}
" COMPILER_SUPPORT_AVX2)
ENDIF()
IF(COMPILER_SUPPORT_SSE42)

9
include/util/tcompression.h
View File

@ -152,15 +152,12 @@ int32_t tsDecompressBigint(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int
// for internal usage
int32_t getWordLength(char type);
#ifdef __AVX2__
int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, char *const output, const char type);
int32_t tsDecompressFloatImpAvx2(const char *input, int32_t nelements, char *output);
int32_t tsDecompressDoubleImpAvx2(const char *input, int32_t nelements, char *output);
#endif
#ifdef __AVX512VL__
void tsDecompressTimestampAvx2(const char *input, int32_t nelements, char *output, bool bigEndian);
void tsDecompressTimestampAvx512(const char *const input, const int32_t nelements, char *const output, bool bigEndian);
#endif
int32_t tsDecompressTimestampAvx2(const char *input, int32_t nelements, char *output, bool bigEndian);
int32_t tsDecompressTimestampAvx512(const char *const input, const int32_t nelements, char *const output,
bool bigEndian);
/*************************************************************************
* REGULAR COMPRESSION 2

1
include/util/tdef.h
View File

@ -41,6 +41,7 @@ extern const int32_t TYPE_BYTES[21];
#define FLOAT_BYTES sizeof(float)
#define DOUBLE_BYTES sizeof(double)
#define POINTER_BYTES sizeof(void *)
#define M256_BYTES 32
#define TSDB_KEYSIZE sizeof(TSKEY)
#define TSDB_NCHAR_SIZE sizeof(TdUcs4)

4
source/libs/function/CMakeLists.txt
View File

@ -1,6 +1,10 @@
aux_source_directory(src FUNCTION_SRC)
aux_source_directory(src/detail FUNCTION_SRC_DETAIL)
list(REMOVE_ITEM FUNCTION_SRC src/udfd.c)
IF(COMPILER_SUPPORT_AVX2)
MESSAGE(STATUS "AVX2 instructions is ACTIVATED")
set_source_files_properties(src/detail/tminmaxavx.c PROPERTIES COMPILE_FLAGS -mavx2)
ENDIF()
add_library(function STATIC ${FUNCTION_SRC} ${FUNCTION_SRC_DETAIL})
target_include_directories(
function

5
source/libs/function/inc/builtinsimpl.h
View File

@ -25,6 +25,11 @@ extern "C" {
#include "functionResInfoInt.h"
int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc, int32_t* nElems);
int32_t i8VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res);
int32_t i16VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res);
int32_t i32VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res);
int32_t floatVectorCmpAVX2(const float* pData, int32_t numOfRows, bool isMinFunc, float* res);
int32_t doubleVectorCmpAVX2(const double* pData, int32_t numOfRows, bool isMinFunc, double* res);
int32_t saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
int32_t updateTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);

247
source/libs/function/src/detail/tminmax.c
View File

@ -72,173 +72,6 @@
#define GET_INVOKE_INTRINSIC_THRESHOLD(_bits, _bytes) ((_bits) / ((_bytes) << 3u))
#ifdef __AVX2__
static void calculateRounds(int32_t numOfRows, int32_t bytes, int32_t* remainder, int32_t* rounds, int32_t* width) {
const int32_t bitWidth = 256;
*width = (bitWidth >> 3u) / bytes;
*remainder = numOfRows % (*width);
*rounds = numOfRows / (*width);
}
#define EXTRACT_MAX_VAL(_first, _sec, _width, _remain, _v) \
__COMPARE_EXTRACT_MAX(0, (_width), (_v), (_first)) \
__COMPARE_EXTRACT_MAX(0, (_remain), (_v), (_sec))
#define EXTRACT_MIN_VAL(_first, _sec, _width, _remain, _v) \
__COMPARE_EXTRACT_MIN(0, (_width), (_v), (_first)) \
__COMPARE_EXTRACT_MIN(0, (_remain), (_v), (_sec))
#define CMP_TYPE_MIN_MAX(type, cmp) \
const type* p = pData; \
__m256i initVal = _mm256_lddqu_si256((__m256i*)p); \
p += width; \
for (int32_t i = 1; i < (rounds); ++i) { \
__m256i next = _mm256_lddqu_si256((__m256i*)p); \
initVal = CMP_FUNC_##cmp##_##type(initVal, next); \
p += width; \
} \
const type* q = (const type*)&initVal; \
type* v = (type*)res; \
EXTRACT_##cmp##_VAL(q, p, width, remain, *v)
static void i8VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res) {
const int8_t* p = pData;
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(int8_t), &remain, &rounds, &width);
#define CMP_FUNC_MIN_int8_t _mm256_min_epi8
#define CMP_FUNC_MAX_int8_t _mm256_max_epi8
#define CMP_FUNC_MIN_uint8_t _mm256_min_epu8
#define CMP_FUNC_MAX_uint8_t _mm256_max_epu8
if (!isMinFunc) { // max function
if (signVal) {
CMP_TYPE_MIN_MAX(int8_t, MAX);
} else {
CMP_TYPE_MIN_MAX(uint8_t, MAX);
}
} else { // min function
if (signVal) {
CMP_TYPE_MIN_MAX(int8_t, MIN);
} else {
CMP_TYPE_MIN_MAX(uint8_t, MIN);
}
}
}
static void i16VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res) {
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(int16_t), &remain, &rounds, &width);
#define CMP_FUNC_MIN_int16_t _mm256_min_epi16
#define CMP_FUNC_MAX_int16_t _mm256_max_epi16
#define CMP_FUNC_MIN_uint16_t _mm256_min_epu16
#define CMP_FUNC_MAX_uint16_t _mm256_max_epu16
if (!isMinFunc) { // max function
if (signVal) {
CMP_TYPE_MIN_MAX(int16_t, MAX);
} else {
CMP_TYPE_MIN_MAX(uint16_t, MAX);
}
} else { // min function
if (signVal) {
CMP_TYPE_MIN_MAX(int16_t, MIN);
} else {
CMP_TYPE_MIN_MAX(uint16_t, MIN);
}
}
}
static void i32VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res) {
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(int32_t), &remain, &rounds, &width);
#define CMP_FUNC_MIN_int32_t _mm256_min_epi32
#define CMP_FUNC_MAX_int32_t _mm256_max_epi32
#define CMP_FUNC_MIN_uint32_t _mm256_min_epu32
#define CMP_FUNC_MAX_uint32_t _mm256_max_epu32
if (!isMinFunc) { // max function
if (signVal) {
CMP_TYPE_MIN_MAX(int32_t, MAX);
} else {
CMP_TYPE_MIN_MAX(uint32_t, MAX);
}
} else { // min function
if (signVal) {
CMP_TYPE_MIN_MAX(int32_t, MIN);
} else {
CMP_TYPE_MIN_MAX(uint32_t, MIN);
}
}
}
static void floatVectorCmpAVX2(const float* pData, int32_t numOfRows, bool isMinFunc, float* res) {
const float* p = pData;
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(float), &remain, &rounds, &width);
__m256 next;
__m256 initVal = _mm256_loadu_ps(p);
p += width;
if (!isMinFunc) { // max function
for (int32_t i = 1; i < rounds; ++i) {
next = _mm256_loadu_ps(p);
initVal = _mm256_max_ps(initVal, next);
p += width;
}
const float* q = (const float*)&initVal;
EXTRACT_MAX_VAL(q, p, width, remain, *res)
} else { // min function
for (int32_t i = 1; i < rounds; ++i) {
next = _mm256_loadu_ps(p);
initVal = _mm256_min_ps(initVal, next);
p += width;
}
const float* q = (const float*)&initVal;
EXTRACT_MIN_VAL(q, p, width, remain, *res)
}
}
static void doubleVectorCmpAVX2(const double* pData, int32_t numOfRows, bool isMinFunc, double* res) {
const double* p = pData;
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(double), &remain, &rounds, &width);
__m256d next;
__m256d initVal = _mm256_loadu_pd(p);
p += width;
if (!isMinFunc) { // max function
for (int32_t i = 1; i < rounds; ++i) {
next = _mm256_loadu_pd(p);
initVal = _mm256_max_pd(initVal, next);
p += width;
}
// let sum up the final results
const double* q = (const double*)&initVal;
EXTRACT_MAX_VAL(q, p, width, remain, *res)
} else { // min function
for (int32_t i = 1; i < rounds; ++i) {
next = _mm256_loadu_pd(p);
initVal = _mm256_min_pd(initVal, next);
p += width;
}
// let sum up the final results
const double* q = (const double*)&initVal;
EXTRACT_MIN_VAL(q, p, width, remain, *res)
}
}
#endif
static int32_t findFirstValPosition(const SColumnInfoData* pCol, int32_t start, int32_t numOfRows, bool isStr) {
int32_t i = start;
@ -255,13 +88,14 @@ static void handleInt8Col(const void* data, int32_t start, int32_t numOfRows, SM
pBuf->v = ((const int8_t*)data)[start];
}
#ifdef __AVX2__
if (tsAVX2Supported && tsSIMDEnable && numOfRows * sizeof(int8_t) >= sizeof(__m256i)) {
i8VectorCmpAVX2(data + start * sizeof(int8_t), numOfRows, isMinFunc, signVal, &pBuf->v);
} else {
#else
if (true) {
#endif
if (tsAVX2Supported && tsSIMDEnable && numOfRows * sizeof(int8_t) >= M256_BYTES) {
int32_t code = i8VectorCmpAVX2(((char*)data) + start * sizeof(int8_t), numOfRows, isMinFunc, signVal, &pBuf->v);
if (code == TSDB_CODE_SUCCESS) {
pBuf->assign = true;
return;
}
}
if (signVal) {
const int8_t* p = (const int8_t*)data;
int8_t* v = (int8_t*)&pBuf->v;
@ -281,7 +115,6 @@ static void handleInt8Col(const void* data, int32_t start, int32_t numOfRows, SM
__COMPARE_EXTRACT_MAX(start, start + numOfRows, *v, p);
}
}
}
pBuf->assign = true;
}
@ -292,13 +125,14 @@ static void handleInt16Col(const void* data, int32_t start, int32_t numOfRows, S
pBuf->v = ((const int16_t*)data)[start];
}
#ifdef __AVX2__
if (tsAVX2Supported && tsSIMDEnable && numOfRows * sizeof(int16_t) >= sizeof(__m256i)) {
i16VectorCmpAVX2(data + start * sizeof(int16_t), numOfRows, isMinFunc, signVal, &pBuf->v);
} else {
#else
if (true) {
#endif
if (tsAVX2Supported && tsSIMDEnable && numOfRows * sizeof(int16_t) >= M256_BYTES) {
int32_t code = i16VectorCmpAVX2(((char*)data) + start * sizeof(int16_t), numOfRows, isMinFunc, signVal, &pBuf->v);
if (code == TSDB_CODE_SUCCESS) {
pBuf->assign = true;
return;
}
}
if (signVal) {
const int16_t* p = (const int16_t*)data;
int16_t* v = (int16_t*)&pBuf->v;
@ -318,7 +152,6 @@ static void handleInt16Col(const void* data, int32_t start, int32_t numOfRows, S
__COMPARE_EXTRACT_MAX(start, start + numOfRows, *v, p);
}
}
}
pBuf->assign = true;
}
@ -329,13 +162,14 @@ static void handleInt32Col(const void* data, int32_t start, int32_t numOfRows, S
pBuf->v = ((const int32_t*)data)[start];
}
#ifdef __AVX2__
if (tsAVX2Supported && tsSIMDEnable && numOfRows * sizeof(int32_t) >= sizeof(__m256i)) {
i32VectorCmpAVX2(data + start * sizeof(int32_t), numOfRows, isMinFunc, signVal, &pBuf->v);
} else {
#else
if (true) {
#endif
if (tsAVX2Supported && tsSIMDEnable && numOfRows * sizeof(int32_t) >= M256_BYTES) {
int32_t code = i32VectorCmpAVX2(((char*)data) + start * sizeof(int32_t), numOfRows, isMinFunc, signVal, &pBuf->v);
if (code == TSDB_CODE_SUCCESS) {
pBuf->assign = true;
return;
}
}
if (signVal) {
const int32_t* p = (const int32_t*)data;
int32_t* v = (int32_t*)&pBuf->v;
@ -355,7 +189,6 @@ static void handleInt32Col(const void* data, int32_t start, int32_t numOfRows, S
__COMPARE_EXTRACT_MAX(start, start + numOfRows, *v, p);
}
}
}
pBuf->assign = true;
}
@ -397,19 +230,19 @@ static void handleFloatCol(SColumnInfoData* pCol, int32_t start, int32_t numOfRo
*val = pData[start];
}
#ifdef __AVX2__
if (tsAVXSupported && tsSIMDEnable && numOfRows * sizeof(float) >= sizeof(__m256i)) {
floatVectorCmpAVX2(pData + start, numOfRows, isMinFunc, val);
} else {
#else
if (true) {
#endif
if (tsAVX2Supported && tsSIMDEnable && numOfRows * sizeof(float) >= M256_BYTES) {
int32_t code = floatVectorCmpAVX2(pData + start, numOfRows, isMinFunc, val);
if (code == TSDB_CODE_SUCCESS) {
pBuf->assign = true;
return;
}
}
if (isMinFunc) { // min
__COMPARE_EXTRACT_MIN(start, start + numOfRows, *val, pData);
} else { // max
__COMPARE_EXTRACT_MAX(start, start + numOfRows, *val, pData);
}
}
pBuf->assign = true;
}
@ -422,19 +255,19 @@ static void handleDoubleCol(SColumnInfoData* pCol, int32_t start, int32_t numOfR
*val = pData[start];
}
#ifdef __AVX2__
if (tsAVXSupported && tsSIMDEnable && numOfRows * sizeof(double) >= sizeof(__m256i)) {
doubleVectorCmpAVX2(pData + start, numOfRows, isMinFunc, val);
} else {
#else
if (true) {
#endif
if (tsAVX2Supported && tsSIMDEnable && numOfRows * sizeof(double) >= M256_BYTES) {
int32_t code = doubleVectorCmpAVX2(pData + start, numOfRows, isMinFunc, val);
if (code == TSDB_CODE_SUCCESS) {
pBuf->assign = true;
return;
}
}
if (isMinFunc) { // min
__COMPARE_EXTRACT_MIN(start, start + numOfRows, *val, pData);
} else { // max
__COMPARE_EXTRACT_MAX(start, start + numOfRows, *val, pData);
}
}
pBuf->assign = true;
}

227
source/libs/function/src/detail/tminmaxavx.c Normal file
View File

@ -0,0 +1,227 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "builtinsimpl.h"
#ifdef __AVX2__
static void calculateRounds(int32_t numOfRows, int32_t bytes, int32_t* remainder, int32_t* rounds, int32_t* width) {
const int32_t bitWidth = 256;
*width = (bitWidth >> 3u) / bytes;
*remainder = numOfRows % (*width);
*rounds = numOfRows / (*width);
}
#define __COMPARE_EXTRACT_MIN(start, end, val, _data) \
for (int32_t i = (start); i < (end); ++i) { \
if ((val) > (_data)[i]) { \
(val) = (_data)[i]; \
} \
}
#define __COMPARE_EXTRACT_MAX(start, end, val, _data) \
for (int32_t i = (start); i < (end); ++i) { \
if ((val) < (_data)[i]) { \
(val) = (_data)[i]; \
} \
}
#define EXTRACT_MAX_VAL(_first, _sec, _width, _remain, _v) \
__COMPARE_EXTRACT_MAX(0, (_width), (_v), (_first)) \
__COMPARE_EXTRACT_MAX(0, (_remain), (_v), (_sec))
#define EXTRACT_MIN_VAL(_first, _sec, _width, _remain, _v) \
__COMPARE_EXTRACT_MIN(0, (_width), (_v), (_first)) \
__COMPARE_EXTRACT_MIN(0, (_remain), (_v), (_sec))
#define CMP_TYPE_MIN_MAX(type, cmp) \
const type* p = pData; \
__m256i initVal = _mm256_lddqu_si256((__m256i*)p); \
p += width; \
for (int32_t i = 1; i < (rounds); ++i) { \
__m256i next = _mm256_lddqu_si256((__m256i*)p); \
initVal = CMP_FUNC_##cmp##_##type(initVal, next); \
p += width; \
} \
const type* q = (const type*)&initVal; \
type* v = (type*)res; \
EXTRACT_##cmp##_VAL(q, p, width, remain, *v)
#endif
int32_t i8VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res) {
#ifdef __AVX2__
const int8_t* p = pData;
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(int8_t), &remain, &rounds, &width);
#define CMP_FUNC_MIN_int8_t _mm256_min_epi8
#define CMP_FUNC_MAX_int8_t _mm256_max_epi8
#define CMP_FUNC_MIN_uint8_t _mm256_min_epu8
#define CMP_FUNC_MAX_uint8_t _mm256_max_epu8
if (!isMinFunc) { // max function
if (signVal) {
CMP_TYPE_MIN_MAX(int8_t, MAX);
} else {
CMP_TYPE_MIN_MAX(uint8_t, MAX);
}
} else { // min function
if (signVal) {
CMP_TYPE_MIN_MAX(int8_t, MIN);
} else {
CMP_TYPE_MIN_MAX(uint8_t, MIN);
}
}
return TSDB_CODE_SUCCESS;
#else
uError("unable run %s without avx2 instructions", __func__);
return TSDB_CODE_OPS_NOT_SUPPORT;
#endif
}
int32_t i16VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res) {
#ifdef __AVX2__
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(int16_t), &remain, &rounds, &width);
#define CMP_FUNC_MIN_int16_t _mm256_min_epi16
#define CMP_FUNC_MAX_int16_t _mm256_max_epi16
#define CMP_FUNC_MIN_uint16_t _mm256_min_epu16
#define CMP_FUNC_MAX_uint16_t _mm256_max_epu16
if (!isMinFunc) { // max function
if (signVal) {
CMP_TYPE_MIN_MAX(int16_t, MAX);
} else {
CMP_TYPE_MIN_MAX(uint16_t, MAX);
}
} else { // min function
if (signVal) {
CMP_TYPE_MIN_MAX(int16_t, MIN);
} else {
CMP_TYPE_MIN_MAX(uint16_t, MIN);
}
}
return TSDB_CODE_SUCCESS;
#else
uError("unable run %s without avx2 instructions", __func__);
return TSDB_CODE_OPS_NOT_SUPPORT;
#endif
}
int32_t i32VectorCmpAVX2(const void* pData, int32_t numOfRows, bool isMinFunc, bool signVal, int64_t* res) {
#ifdef __AVX2__
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(int32_t), &remain, &rounds, &width);
#define CMP_FUNC_MIN_int32_t _mm256_min_epi32
#define CMP_FUNC_MAX_int32_t _mm256_max_epi32
#define CMP_FUNC_MIN_uint32_t _mm256_min_epu32
#define CMP_FUNC_MAX_uint32_t _mm256_max_epu32
if (!isMinFunc) { // max function
if (signVal) {
CMP_TYPE_MIN_MAX(int32_t, MAX);
} else {
CMP_TYPE_MIN_MAX(uint32_t, MAX);
}
} else { // min function
if (signVal) {
CMP_TYPE_MIN_MAX(int32_t, MIN);
} else {
CMP_TYPE_MIN_MAX(uint32_t, MIN);
}
}
return TSDB_CODE_SUCCESS;
#else
uError("unable run %s without avx2 instructions", __func__);
return TSDB_CODE_OPS_NOT_SUPPORT;
#endif
}
int32_t floatVectorCmpAVX2(const float* pData, int32_t numOfRows, bool isMinFunc, float* res) {
#ifdef __AVX2__
const float* p = pData;
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(float), &remain, &rounds, &width);
__m256 next;
__m256 initVal = _mm256_loadu_ps(p);
p += width;
if (!isMinFunc) { // max function
for (int32_t i = 1; i < rounds; ++i) {
next = _mm256_loadu_ps(p);
initVal = _mm256_max_ps(initVal, next);
p += width;
}
const float* q = (const float*)&initVal;
EXTRACT_MAX_VAL(q, p, width, remain, *res)
} else { // min function
for (int32_t i = 1; i < rounds; ++i) {
next = _mm256_loadu_ps(p);
initVal = _mm256_min_ps(initVal, next);
p += width;
}
const float* q = (const float*)&initVal;
EXTRACT_MIN_VAL(q, p, width, remain, *res)
}
return TSDB_CODE_SUCCESS;
#else
uError("unable run %s without avx2 instructions", __func__);
return TSDB_CODE_OPS_NOT_SUPPORT;
#endif
}
int32_t doubleVectorCmpAVX2(const double* pData, int32_t numOfRows, bool isMinFunc, double* res) {
#ifdef __AVX2__
const double* p = pData;
int32_t width, remain, rounds;
calculateRounds(numOfRows, sizeof(double), &remain, &rounds, &width);
__m256d next;
__m256d initVal = _mm256_loadu_pd(p);
p += width;
if (!isMinFunc) { // max function
for (int32_t i = 1; i < rounds; ++i) {
next = _mm256_loadu_pd(p);
initVal = _mm256_max_pd(initVal, next);
p += width;
}
// let sum up the final results
const double* q = (const double*)&initVal;
EXTRACT_MAX_VAL(q, p, width, remain, *res)
} else { // min function
for (int32_t i = 1; i < rounds; ++i) {
next = _mm256_loadu_pd(p);
initVal = _mm256_min_pd(initVal, next);
p += width;
}
// let sum up the final results
const double* q = (const double*)&initVal;
EXTRACT_MIN_VAL(q, p, width, remain, *res)
}
return TSDB_CODE_SUCCESS;
#else
uError("unable run %s without avx2 instructions", __func__);
return TSDB_CODE_OPS_NOT_SUPPORT;
#endif
}

1
source/os/src/osEnv.c
View File

@ -37,7 +37,6 @@ float tsNumOfCores = 0;
int64_t tsTotalMemoryKB = 0;
char *tsProcPath = NULL;
char tsSIMDEnable = 1;
char tsAVX512Enable = 0;
char tsSSE42Supported = 0;
char tsAVXSupported = 0;

4
source/util/CMakeLists.txt
View File

@ -1,5 +1,9 @@
configure_file("${CMAKE_CURRENT_SOURCE_DIR}/src/version.c.in" "${CMAKE_CURRENT_SOURCE_DIR}/src/version.c")
aux_source_directory(src UTIL_SRC)
IF(COMPILER_SUPPORT_AVX2)
MESSAGE(STATUS "AVX2 instructions is ACTIVATED")
set_source_files_properties(src/tdecompressavx.c PROPERTIES COMPILE_FLAGS -mavx2)
ENDIF()
add_library(util STATIC ${UTIL_SRC})
if(DEFINED GRANT_CFG_INCLUDE_DIR)

53
source/util/src/tcompression.c
View File

@ -471,12 +471,12 @@ int32_t tsDecompressINTImp(const char *const input, const int32_t nelements, cha
return nelements * word_length;
}
#ifdef __AVX512F__
if (tsSIMDEnable && tsAVX512Enable && tsAVX512Supported) {
tsDecompressIntImpl_Hw(input, nelements, output, type);
return nelements * word_length;
int32_t cnt = tsDecompressIntImpl_Hw(input, nelements, output, type);
if (cnt >= 0) {
return cnt;
}
}
#endif
// Selector value: 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
char bit_per_integer[] = {0, 0, 1, 2, 3, 4, 5, 6, 7, 8, 10, 12, 15, 20, 30, 60};
@ -867,12 +867,12 @@ int32_t tsDecompressTimestampImp(const char *const input, const int32_t nelement
memcpy(output, input + 1, nelements * longBytes);
return nelements * longBytes;
} else if (input[0] == 1) { // Decompress
#ifdef __AVX512VL__
if (tsSIMDEnable && tsAVX512Enable && tsAVX512Supported) {
tsDecompressTimestampAvx512(const char *const input, const int32_t nelements, char *const output, bool bigEndian);
return nelements * longBytes;
int32_t cnt = tsDecompressTimestampAvx512(input, nelements, output, false);
if (cnt >= 0) {
return cnt;
}
}
#endif
int64_t *ostream = (int64_t *)output;
@ -1103,13 +1103,14 @@ int32_t tsDecompressDoubleImp(const char *const input, int32_t ninput, const int
return nelements * DOUBLE_BYTES;
}
#ifdef __AVX2__
// use AVX2 implementation when allowed and the compression ratio is not high
double compressRatio = 1.0 * nelements * DOUBLE_BYTES / ninput;
if (tsSIMDEnable && tsAVX2Supported && compressRatio < 2) {
return tsDecompressDoubleImpAvx2(input + 1, nelements, output);
int32_t cnt = tsDecompressDoubleImpAvx2(input + 1, nelements, output);
if (cnt >= 0) {
return cnt;
}
}
#endif
// use implementation without SIMD instructions by default
return tsDecompressDoubleImpHelper(input + 1, nelements, output);
@ -1257,13 +1258,14 @@ int32_t tsDecompressFloatImp(const char *const input, int32_t ninput, const int3
return nelements * FLOAT_BYTES;
}
#ifdef __AVX2__
// use AVX2 implementation when allowed and the compression ratio is not high
double compressRatio = 1.0 * nelements * FLOAT_BYTES / ninput;
if (tsSIMDEnable && tsAVX2Supported && compressRatio < 2) {
return tsDecompressFloatImpAvx2(input + 1, nelements, output);
int32_t cnt = tsDecompressFloatImpAvx2(input + 1, nelements, output);
if (cnt >= 0) {
return cnt;
}
}
#endif
// use implementation without SIMD instructions by default
return tsDecompressFloatImpHelper(input + 1, nelements, output);
@ -1883,3 +1885,26 @@ int8_t tUpdateCompress(uint32_t oldCmpr, uint32_t newCmpr, uint8_t l2Disabled, u
return update;
}
int32_t getWordLength(char type) {
int32_t wordLength = 0;
switch (type) {
case TSDB_DATA_TYPE_BIGINT:
wordLength = LONG_BYTES;
break;
case TSDB_DATA_TYPE_INT:
wordLength = INT_BYTES;
break;
case TSDB_DATA_TYPE_SMALLINT:
wordLength = SHORT_BYTES;
break;
case TSDB_DATA_TYPE_TINYINT:
wordLength = CHAR_BYTES;
break;
default:
uError("Invalid decompress integer type:%d", type);
return TSDB_CODE_INVALID_PARA;
}
return wordLength;
}

116
source/util/src/tdecompress.c → source/util/src/tdecompressavx.c
View File

@ -13,35 +13,16 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "tcompression.h"
#include "ttypes.h"
int32_t getWordLength(char type) {
int32_t wordLength = 0;
switch (type) {
case TSDB_DATA_TYPE_BIGINT:
wordLength = LONG_BYTES;
break;
case TSDB_DATA_TYPE_INT:
wordLength = INT_BYTES;
break;
case TSDB_DATA_TYPE_SMALLINT:
wordLength = SHORT_BYTES;
break;
case TSDB_DATA_TYPE_TINYINT:
wordLength = CHAR_BYTES;
break;
default:
uError("Invalid decompress integer type:%d", type);
return TSDB_CODE_INVALID_PARA;
}
return wordLength;
}
#ifdef __AVX2__
char tsSIMDEnable = 1;
#else
char tsSIMDEnable = 0;
#endif
int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, char *const output, const char type) {
#ifdef __AVX2__
int32_t word_length = getWordLength(type);
// Selector value: 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
@ -75,12 +56,12 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements,
int32_t batch = 0;
int32_t remain = 0;
if (tsSIMDEnable && tsAVX512Supported && tsAVX512Enable) {
#if __AVX512F__
#ifdef __AVX512F__
batch = num >> 3;
remain = num & 0x07;
#endif
} else if (tsSIMDEnable && tsAVX2Supported) {
#if __AVX2__
#ifdef __AVX2__
batch = num >> 2;
remain = num & 0x03;
#endif
@ -88,7 +69,7 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements,
if (selector == 0 || selector == 1) {
if (tsSIMDEnable && tsAVX512Supported && tsAVX512Enable) {
#if __AVX512F__
#ifdef __AVX512F__
for (int32_t i = 0; i < batch; ++i) {
__m512i prev = _mm512_set1_epi64(prevValue);
_mm512_storeu_si512((__m512i *)&p[_pos], prev);
@ -117,7 +98,7 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements,
}
} else {
if (tsSIMDEnable && tsAVX512Supported && tsAVX512Enable) {
#if __AVX512F__
#ifdef __AVX512F__
__m512i sum_mask1 = _mm512_set_epi64(6, 6, 4, 4, 2, 2, 0, 0);
__m512i sum_mask2 = _mm512_set_epi64(5, 5, 5, 5, 1, 1, 1, 1);
__m512i sum_mask3 = _mm512_set_epi64(3, 3, 3, 3, 3, 3, 3, 3);
@ -310,10 +291,13 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements,
}
return nelements * word_length;
#else
uError("unable run %s without avx2 instructions", __func__);
return -1;
#endif
}
#define M256_BYTES sizeof(__m256i)
#ifdef __AVX2__
FORCE_INLINE __m256i decodeFloatAvx2(const char *data, const char *flag) {
__m256i dataVec = _mm256_load_si256((__m256i *)data);
__m256i flagVec = _mm256_load_si256((__m256i *)flag);
@ -332,7 +316,27 @@ FORCE_INLINE __m256i decodeFloatAvx2(const char *data, const char *flag) {
return diffVec;
}
FORCE_INLINE __m256i decodeDoubleAvx2(const char *data, const char *flag) {
__m256i dataVec = _mm256_load_si256((__m256i *)data);
__m256i flagVec = _mm256_load_si256((__m256i *)flag);
__m256i k7 = _mm256_set1_epi64x(7);
__m256i lopart = _mm256_set_epi64x(0, -1, 0, -1);
__m256i hipart = _mm256_set_epi64x(-1, 0, -1, 0);
__m256i trTail = _mm256_cmpgt_epi64(flagVec, k7);
__m256i trHead = _mm256_andnot_si256(trTail, _mm256_set1_epi64x(-1));
__m256i shiftVec = _mm256_slli_epi64(_mm256_sub_epi64(k7, _mm256_and_si256(flagVec, k7)), 3);
__m256i maskVec = hipart;
__m256i diffVec = _mm256_sllv_epi64(dataVec, _mm256_and_si256(shiftVec, maskVec));
maskVec = _mm256_or_si256(trHead, lopart);
diffVec = _mm256_srlv_epi64(diffVec, _mm256_and_si256(shiftVec, maskVec));
maskVec = _mm256_and_si256(trTail, lopart);
diffVec = _mm256_sllv_epi64(diffVec, _mm256_and_si256(shiftVec, maskVec));
return diffVec;
}
#endif
int32_t tsDecompressFloatImpAvx2(const char *input, int32_t nelements, char *output) {
#ifdef __AVX2__
// Allocate memory-aligned buffer
char buf[M256_BYTES * 3];
memset(buf, 0, sizeof(buf));
@ -380,27 +384,14 @@ int32_t tsDecompressFloatImpAvx2(const char *input, int32_t nelements, char *out
out += idx * FLOAT_BYTES;
}
return (int32_t)(out - output);
}
FORCE_INLINE __m256i decodeDoubleAvx2(const char *data, const char *flag) {
__m256i dataVec = _mm256_load_si256((__m256i *)data);
__m256i flagVec = _mm256_load_si256((__m256i *)flag);
__m256i k7 = _mm256_set1_epi64x(7);
__m256i lopart = _mm256_set_epi64x(0, -1, 0, -1);
__m256i hipart = _mm256_set_epi64x(-1, 0, -1, 0);
__m256i trTail = _mm256_cmpgt_epi64(flagVec, k7);
__m256i trHead = _mm256_andnot_si256(trTail, _mm256_set1_epi64x(-1));
__m256i shiftVec = _mm256_slli_epi64(_mm256_sub_epi64(k7, _mm256_and_si256(flagVec, k7)), 3);
__m256i maskVec = hipart;
__m256i diffVec = _mm256_sllv_epi64(dataVec, _mm256_and_si256(shiftVec, maskVec));
maskVec = _mm256_or_si256(trHead, lopart);
diffVec = _mm256_srlv_epi64(diffVec, _mm256_and_si256(shiftVec, maskVec));
maskVec = _mm256_and_si256(trTail, lopart);
diffVec = _mm256_sllv_epi64(diffVec, _mm256_and_si256(shiftVec, maskVec));
return diffVec;
#else
uError("unable run %s without avx2 instructions", __func__);
return -1;
#endif
}
int32_t tsDecompressDoubleImpAvx2(const char *input, const int32_t nelements, char *const output) {
#ifdef __AVX2__
// Allocate memory-aligned buffer
char buf[M256_BYTES * 3];
memset(buf, 0, sizeof(buf));
@ -448,12 +439,15 @@ int32_t tsDecompressDoubleImpAvx2(const char *input, const int32_t nelements, ch
out += idx * DOUBLE_BYTES;
}
return (int32_t)(out - output);
}
#else
uError("unable run %s without avx2 instructions", __func__);
return -1;
#endif
}
#if __AVX512VL__
// decode two timestamps in one loop.
void tsDecompressTimestampAvx2(const char *const input, const int32_t nelements, char *const output, bool bigEndian) {
int32_t tsDecompressTimestampAvx2(const char *const input, const int32_t nelements, char *const output,
bool bigEndian) {
#ifdef __AVX512VL__
int64_t *ostream = (int64_t *)output;
int32_t ipos = 1, opos = 0;
@ -588,11 +582,16 @@ void tsDecompressTimestampAvx2(const char *const input, const int32_t nelements,
ostream[opos++] = prevVal[1] + prevDeltaX;
}
}
return;
return opos;
#else
uError("unable run %s without avx512 instructions", __func__);
return -1;
#endif
}
void tsDecompressTimestampAvx512(const char *const input, const int32_t nelements, char *const output,
int32_t tsDecompressTimestampAvx512(const char *const input, const int32_t nelements, char *const output,
bool UNUSED_PARAM(bigEndian)) {
#ifdef __AVX512VL__
int64_t *ostream = (int64_t *)output;
int32_t ipos = 1, opos = 0;
@ -700,6 +699,9 @@ void tsDecompressTimestampAvx512(const char *const input, const int32_t nelement
}
}
return;
}
return opos;
#else
uError("unable run %s without avx512 instructions", __func__);
return -1;
#endif
}