Optimize string2ll performance using AVX512 (#1944)

### Description

This pull request aims to optimize the `string2ll` function using AVX512
instructions to enhance performance for string-to-integer conversions.
The existing scalar version of the function was identified as a
bottleneck in certain scenarios, and this optimization seeks to address
that.

### Changes

- Introduced string2llAVX512 function in util.c to utilize AVX512 SIMD
instructions.
- Updated config.h to define ATTRIBUTE_TARGET_AVX512 for AVX512 support.
- Replaced calls to `lpStringToInt64` with `string2ll` for consistency
and performance improvement.
- Removed redundant code in listpack.c and optimized integer encoding logic.

### Performance Boost

The corresponding list commands can gain up to **~19%** performance
improvement, the string length is 4 bytes. The longer the string, the
greater the performance boost.

### Ref

1. http://0x80.pl/notesen/2018-04-19-simd-parsing-int-sequences.html#toc-entry-1
2. https://lemire.me/blog/2023/09/22/parsing-integers-quickly-with-avx-512/

---------

Signed-off-by: Lipeng Zhu <lipeng.zhu@intel.com>
Signed-off-by: Lipeng Zhu <zhu.lipeng@outlook.com>
Co-authored-by: Wangyang Guo <wangyang.guo@intel.com>
Co-authored-by: Viktor Söderqvist <viktor.soderqvist@est.tech>

Author: 3 people

src/bitops.c

@@ -29,7 +29,7 @@
  */
 
 #include "server.h"
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
 /* Define __MM_MALLOC_H to prevent importing the memory aligned
  * allocation functions, which we don't use. */
 #define __MM_MALLOC_H
@@ -48,7 +48,7 @@ static const unsigned char bitsinbyte[256] = {
     5, 5, 6, 5, 6, 6, 7, 2, 3, 3, 4, 3, 4, 4, 5, 3, 4, 4, 5, 4, 5, 5, 6, 3, 4, 4, 5, 4, 5, 5, 6, 4, 5, 5, 6, 5, 6,
     6, 7, 3, 4, 4, 5, 4, 5, 5, 6, 4, 5, 5, 6, 5, 6, 6, 7, 4, 5, 5, 6, 5, 6, 6, 7, 5, 6, 6, 7, 6, 7, 7, 8};
 
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
 /* The SIMD version of popcount enhances performance through parallel lookup tables which is based on the following article:
  * https://arxiv.org/pdf/1611.07612 */
 ATTRIBUTE_TARGET_AVX2
@@ -191,7 +191,7 @@ long long popcountScalar(void *s, long count) {
  * 'count' bytes. The implementation of this function is required to
  * work with an input string length up to 512 MB or more (server.proto_max_bulk_len) */
 long long serverPopcount(void *s, long count) {
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
     /* If length of s >= 256 bits and the CPU supports AVX2,
      * we prefer to use the SIMD version */
     if (count >= 32) {

src/config.h

@@ -374,17 +374,19 @@ void setcpuaffinity(const char *cpulist);
 #define valkey_prefetch(addr) ((void)(addr))
 #endif
 
-/* Check if we can compile AVX2 code */
-#if defined(__x86_64__) && ((defined(__GNUC__) && __GNUC__ >= 5) || (defined(__clang__) && __clang_major__ >= 4))
-#if defined(__has_attribute) && __has_attribute(target)
-#define HAVE_AVX2
-#endif
+/* Check if we can compile SIMD code */
+#if defined(__x86_64__) && ((defined(__GNUC__) && __GNUC__ >= 5) || (defined(__clang__) && __clang_major__ >= 4)) && defined(__has_attribute) && __has_attribute(target)
+#define HAVE_X86_SIMD 1
+#else
+#define HAVE_X86_SIMD 0
 #endif
 
-#if defined(HAVE_AVX2)
+#if HAVE_X86_SIMD
 #define ATTRIBUTE_TARGET_AVX2 __attribute__((target("avx2")))
+#define ATTRIBUTE_TARGET_AVX512 __attribute__((target("avx512f,avx512bw,avx512vl")))
 #else
 #define ATTRIBUTE_TARGET_AVX2
+#define ATTRIBUTE_TARGET_AVX512
 #endif
 
 #endif

src/hyperloglog.c

@@ -40,7 +40,7 @@
 #include <stdint.h>
 #include <math.h>
 
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
 /* Define __MM_MALLOC_H to prevent importing the memory aligned
  * allocation functions, which we don't use. */
 #define __MM_MALLOC_H
@@ -220,7 +220,7 @@ struct hllhdr {
 
 static char *invalid_hll_err = "-INVALIDOBJ Corrupted HLL object detected";
 
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
 static int simd_enabled = 1;
 #define HLL_USE_AVX2 (simd_enabled && __builtin_cpu_supports("avx2"))
 #else
@@ -1083,7 +1083,7 @@ int hllAdd(robj *o, unsigned char *ele, size_t elesize) {
     }
 }
 
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
 /* A specialized version of hllMergeDense, optimized for default configurations.
  *
  * Requirements:
@@ -1195,7 +1195,7 @@ void hllMergeDenseAVX2(uint8_t *reg_raw, const uint8_t *reg_dense) {
 
 /* Merge dense-encoded registers to raw registers array. */
 void hllMergeDense(uint8_t *reg_raw, const uint8_t *reg_dense) {
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
     if (HLL_REGISTERS == 16384 && HLL_BITS == 6) {
         if (HLL_USE_AVX2) {
             hllMergeDenseAVX2(reg_raw, reg_dense);
@@ -1258,7 +1258,7 @@ int hllMerge(uint8_t *max, robj *hll) {
     return C_OK;
 }
 
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
 /* A specialized version of hllDenseCompress, optimized for default configurations.
  *
  * Requirements:
@@ -1359,7 +1359,7 @@ void hllDenseCompressAVX2(uint8_t *reg_dense, const uint8_t *reg_raw) {
 
 /* Compress raw registers to dense representation. */
 void hllDenseCompress(uint8_t *reg_dense, const uint8_t *reg_raw) {
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
     if (HLL_REGISTERS == 16384 && HLL_BITS == 6) {
         if (HLL_USE_AVX2) {
             hllDenseCompressAVX2(reg_dense, reg_raw);
@@ -1770,11 +1770,11 @@ void pfdebugCommand(client *c) {
         if (c->argc != 3) goto arityerr;
 
         if (!strcasecmp(c->argv[2]->ptr, "on")) {
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
             simd_enabled = 1;
 #endif
         } else if (!strcasecmp(c->argv[2]->ptr, "off")) {
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
             simd_enabled = 0;
 #endif
         } else {

src/listpack.c

@@ -146,91 +146,6 @@ int lpSafeToAdd(unsigned char *lp, size_t add) {
     return 1;
 }
 
-/* Convert a string into a signed 64 bit integer.
- * The function returns 1 if the string could be parsed into a (non-overflowing)
- * signed 64 bit int, 0 otherwise. The 'value' will be set to the parsed value
- * when the function returns success.
- *
- * Note that this function demands that the string strictly represents
- * a int64 value: no spaces or other characters before or after the string
- * representing the number are accepted, nor zeroes at the start if not
- * for the string "0" representing the zero number.
- *
- * Because of its strictness, it is safe to use this function to check if
- * you can convert a string into a long long, and obtain back the string
- * from the number without any loss in the string representation. *
- *
- * -----------------------------------------------------------------------------
- *
- * Credits: this function was adapted from the Redis OSS source code, file
- * "utils.c", function string2ll(), and is copyright:
- *
- * Copyright(C) 2011, Pieter Noordhuis
- * Copyright(C) 2011, Redis Ltd.
- *
- * The function is released under the BSD 3-clause license.
- */
-int lpStringToInt64(const char *s, unsigned long slen, int64_t *value) {
-    const char *p = s;
-    unsigned long plen = 0;
-    int negative = 0;
-    uint64_t v;
-
-    /* Abort if length indicates this cannot possibly be an int */
-    if (slen == 0 || slen >= LONG_STR_SIZE) return 0;
-
-    /* Special case: first and only digit is 0. */
-    if (slen == 1 && p[0] == '0') {
-        if (value != NULL) *value = 0;
-        return 1;
-    }
-
-    if (p[0] == '-') {
-        negative = 1;
-        p++;
-        plen++;
-
-        /* Abort on only a negative sign. */
-        if (plen == slen) return 0;
-    }
-
-    /* First digit should be 1-9, otherwise the string should just be 0. */
-    if (p[0] >= '1' && p[0] <= '9') {
-        v = p[0] - '0';
-        p++;
-        plen++;
-    } else {
-        return 0;
-    }
-
-    while (plen < slen && p[0] >= '0' && p[0] <= '9') {
-        if (v > (UINT64_MAX / 10)) /* Overflow. */
-            return 0;
-        v *= 10;
-
-        if (v > (UINT64_MAX - (p[0] - '0'))) /* Overflow. */
-            return 0;
-        v += p[0] - '0';
-
-        p++;
-        plen++;
-    }
-
-    /* Return if not all bytes were used. */
-    if (plen < slen) return 0;
-
-    if (negative) {
-        if (v > ((uint64_t)(-(INT64_MIN + 1)) + 1)) /* Overflow. */
-            return 0;
-        if (value != NULL) *value = -v;
-    } else {
-        if (v > INT64_MAX) /* Overflow. */
-            return 0;
-        if (value != NULL) *value = v;
-    }
-    return 1;
-}
-
 /* Create a new, empty listpack.
  * On success the new listpack is returned, otherwise an error is returned.
  * Pre-allocate at least `capacity` bytes of memory,
@@ -331,7 +246,7 @@ static inline void lpEncodeIntegerGetType(int64_t v, unsigned char *intenc, uint
  * in order to be represented. */
 static inline int lpEncodeGetType(unsigned char *ele, uint32_t size, unsigned char *intenc, uint64_t *enclen) {
     int64_t v;
-    if (lpStringToInt64((const char *)ele, size, &v)) {
+    if (string2ll((const char *)ele, size, (long long *)&v)) {
         lpEncodeIntegerGetType(v, intenc, enclen);
         return LP_ENCODING_INT;
     } else {
@@ -704,7 +619,7 @@ unsigned char *lpFind(unsigned char *lp, unsigned char *p, unsigned char *s, uin
                     /* If the entry can be encoded as integer we set it to
                      * 1, else set it to UCHAR_MAX, so that we don't retry
                      * again the next time. */
-                    if (slen >= 32 || slen == 0 || !lpStringToInt64((const char *)s, slen, &vll)) {
+                    if (slen >= 32 || slen == 0 || !string2ll((const char *)s, slen, (long long *)&vll)) {
                         vencoding = UCHAR_MAX;
                     } else {
                         vencoding = 1;
@@ -1373,11 +1288,11 @@ unsigned int lpCompare(unsigned char *p, unsigned char *s, uint32_t slen) {
     if (value) {
         return (slen == sz) && memcmp(value, s, slen) == 0;
     } else {
-        /* We use lpStringToInt64() to get an integer representation of the
+        /* We use string2ll() to get an integer representation of the
          * string 's' and compare it to 'sval', it's much faster than convert
          * integer to string and comparing. */
         int64_t sval;
-        if (lpStringToInt64((const char *)s, slen, &sval)) return sz == sval;
+        if (string2ll((const char *)s, slen, (long long *)&sval)) return sz == sval;
     }
 
     return 0;

src/unit/test_bitops.c

@@ -7,7 +7,7 @@
 #include "../zmalloc.h"
 
 extern long long popcountScalar(void *s, long count);
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
 extern long long popcountAVX2(void *s, long count);
 #endif
 
@@ -36,7 +36,7 @@ static int test_case(const char *msg, int size) {
         long long expect = bitcount(buf, size);
         long long ret_scalar = popcountScalar(buf, size);
         TEST_ASSERT_MESSAGE(msg, expect == ret_scalar);
-#ifdef HAVE_AVX2
+#if HAVE_X86_SIMD
         long long ret_avx2 = popcountAVX2(buf, size);
         TEST_ASSERT_MESSAGE(msg, expect == ret_avx2);
 #endif

src/unit/test_util.c

@@ -69,6 +69,9 @@ int test_string2ll(int argc, char **argv, int flags) {
     valkey_strlcpy(buf, "9223372036854775808", sizeof(buf)); /* overflow */
     TEST_ASSERT(string2ll(buf, strlen(buf), &v) == 0);
 
+    valkey_strlcpy(buf, "18446744073709551615", sizeof(buf)); /* overflow */
+    TEST_ASSERT(string2ll(buf, strlen(buf), &v) == 0);
+
     return 0;
 }