I wrote
> On Mon, Jul 11, 2022 at 11:07 PM Andres Freund <and...@anarazel.de> wrote:
>
> > I wonder if we can add a somewhat more general function for scanning until
> > some characters are found using SIMD? There's plenty other places that could
> > be useful.
>
> In simple cases, we could possibly abstract the entire loop. With this
> particular case, I imagine the most approachable way to write the loop would
> be a bit more low-level:
>
> while (p < end - VECTOR_WIDTH &&
> !vector_has_byte(p, '\\') &&
> !vector_has_byte(p, '"') &&
> vector_min_byte(p, 0x20))
> p += VECTOR_WIDTH
>
> I wonder if we'd lose a bit of efficiency here by not accumulating set bits
> from the three conditions, but it's worth trying.
The attached implements the above, more or less, using new pg_lfind8()
and pg_lfind8_le(), which in turn are based on helper functions that
act on a single vector. The pg_lfind* functions have regression tests,
but I haven't done the same for json yet. I went the extra step to use
bit-twiddling for non-SSE builds using uint64 as a "vector", which
still gives a pretty good boost (test below, min of 3):
master:
356ms
v5:
259ms
v5 disable SSE:
288ms
It still needs a bit of polishing and testing, but I think it's a good
workout for abstracting SIMD out of the way.
-------------
test:
DROP TABLE IF EXISTS long_json_as_text;
CREATE TABLE long_json_as_text AS
with long as (
select repeat(description, 11)
from pg_description
)
select (select json_agg(row_to_json(long))::text as t from long) from
generate_series(1, 100);
VACUUM FREEZE long_json_as_text;
select 1 from long_json_as_text where t::json is null; -- from Andrew upthread
--
John Naylor
EDB: http://www.enterprisedb.com
diff --git a/src/common/jsonapi.c b/src/common/jsonapi.c
index fefd1d24d9..1f9eb134e8 100644
--- a/src/common/jsonapi.c
+++ b/src/common/jsonapi.c
@@ -19,6 +19,7 @@
#include "common/jsonapi.h"
#include "mb/pg_wchar.h"
+#include "port/pg_lfind.h"
#ifndef FRONTEND
#include "miscadmin.h"
@@ -844,7 +845,7 @@ json_lex_string(JsonLexContext *lex)
}
else
{
- char *p;
+ char *p = s;
if (hi_surrogate != -1)
return JSON_UNICODE_LOW_SURROGATE;
@@ -853,7 +854,13 @@ json_lex_string(JsonLexContext *lex)
* Skip to the first byte that requires special handling, so we
* can batch calls to appendBinaryStringInfo.
*/
- for (p = s; p < end; p++)
+ while (p < end - SIZEOF_VECTOR &&
+ !pg_lfind8('\\', (unsigned char*) p, SIZEOF_VECTOR) &&
+ !pg_lfind8('"', (unsigned char*) p, SIZEOF_VECTOR) &&
+ !pg_lfind8_le(0x1F, (unsigned char*) p, SIZEOF_VECTOR))
+ p += SIZEOF_VECTOR;
+
+ for (; p < end; p++)
{
if (*p == '\\' || *p == '"')
break;
diff --git a/src/include/port/pg_lfind.h b/src/include/port/pg_lfind.h
index fb125977b2..e090ea6ac3 100644
--- a/src/include/port/pg_lfind.h
+++ b/src/include/port/pg_lfind.h
@@ -1,7 +1,7 @@
/*-------------------------------------------------------------------------
*
* pg_lfind.h
- * Optimized linear search routines.
+ * Optimized linear search routines using SIMD intrinsics where available
*
* Copyright (c) 2022, PostgreSQL Global Development Group
*
@@ -15,6 +15,76 @@
#include "port/simd.h"
+/*
+ * pg_lfind8
+ *
+ * Return true if there is an element in 'base' that equals 'key', otherwise
+ * return false.
+ */
+static inline bool
+pg_lfind8(uint8 key, uint8 *base, uint32 nelem)
+{
+ uint32 i;
+ /* round down to multiple of vector length */
+ uint32 iterations = nelem & ~(SIZEOF_VECTOR - 1);
+ TYPEOF_VECTOR chunk;
+
+ for (i = 0; i < iterations; i += SIZEOF_VECTOR)
+ {
+#ifdef USE_SSE2
+ chunk = _mm_loadu_si128((const __m128i *) &base[i]);
+#else
+ memcpy(&chunk, &base[i], sizeof(chunk));
+#endif /* USE_SSE2 */
+ if (vector_eq_byte(chunk, key))
+ return true;
+ }
+
+ /* Process the remaining elements one at a time. */
+ for (; i < nelem; i++)
+ {
+ if (key == base[i])
+ return true;
+ }
+
+ return false;
+}
+
+/*
+ * pg_lfind8
+ *
+ * Return true if there is an element in 'base' that is less than or equal to 'key', otherwise
+ * return false.
+ */
+static inline bool
+pg_lfind8_le(uint8 key, uint8 *base, uint32 nelem)
+{
+ uint32 i;
+ /* round down to multiple of vector length */
+ uint32 iterations = nelem & ~(SIZEOF_VECTOR - 1);
+ TYPEOF_VECTOR chunk;
+
+ for (i = 0; i < iterations; i += SIZEOF_VECTOR)
+ {
+#ifdef USE_SSE2
+ chunk = _mm_loadu_si128((const __m128i *) &base[i]);
+#else
+ memcpy(&chunk, &base[i], sizeof(chunk));
+#endif /* USE_SSE2 */
+ if (vector_le_byte(chunk, key))
+ return true;
+ }
+
+ /* Process the remaining elements one at a time. */
+ for (; i < nelem; i++)
+ {
+ if (base[i] <= key)
+ return true;
+ }
+
+ return false;
+}
+
/*
* pg_lfind32
*
@@ -26,7 +96,6 @@ pg_lfind32(uint32 key, uint32 *base, uint32 nelem)
{
uint32 i = 0;
- /* Use SIMD intrinsics where available. */
#ifdef USE_SSE2
/*
diff --git a/src/include/port/simd.h b/src/include/port/simd.h
index a571e79f57..0185bc2ae0 100644
--- a/src/include/port/simd.h
+++ b/src/include/port/simd.h
@@ -25,6 +25,125 @@
#if (defined(__x86_64__) || defined(_M_AMD64))
#include <emmintrin.h>
#define USE_SSE2
+#define TYPEOF_VECTOR __m128i
+
+#else
+#define TYPEOF_VECTOR uint64
+#endif /* (defined(__x86_64__) || defined(_M_AMD64)) */
+
+#define SIZEOF_VECTOR sizeof(TYPEOF_VECTOR)
+
+/* return a vector with all bytes set to c */
+static inline TYPEOF_VECTOR
+vector_broadcast(const uint8 c)
+{
+#ifdef USE_SSE2
+ return _mm_set1_epi8(c);
+#else
+ return ~UINT64CONST(0) / 0xFF * c;
#endif
+}
+
+/* return true if any bytes in the vector are zero */
+static inline bool
+vector_has_zero(const TYPEOF_VECTOR v)
+{
+#ifdef USE_SSE2
+ return _mm_movemask_epi8(_mm_cmpeq_epi8(v, _mm_setzero_si128()));
+#else
+ /* from https://graphics.stanford.edu/~seander/bithacks.html#ZeroInWord */
+ return (v - vector_broadcast(0x01)) & ~v & vector_broadcast(0x80);
+#endif
+}
+
+static inline bool
+vector_eq_byte(const TYPEOF_VECTOR v, const uint8 c)
+{
+ bool result;
+
+ /* pre-compute the result for assert checking */
+#ifdef USE_ASSERT_CHECKING
+ bool assert_result = false;
+ const char* s = (const char*) &v;
+
+ for (int j = 0; j < SIZEOF_VECTOR; j++)
+ {
+ if (s[j] == c)
+ {
+ assert_result = true;
+ break;
+ }
+ }
+#endif /* USE_ASSERT_CHECKING */
+
+#ifdef USE_SSE2
+ result = _mm_movemask_epi8(_mm_cmpeq_epi8(v, vector_broadcast(c)));
+#else
+ /* any bytes in v equal to c will evaluate to zero via XOR */
+ result = vector_has_zero(v ^ vector_broadcast(c));
+#endif /* USE_SSE2 */
+
+#ifdef USE_ASSERT_CHECKING
+ Assert(assert_result == result);
+#endif /* USE_ASSERT_CHECKING */
+
+ return result;
+}
+
+static inline bool
+vector_le_byte(const TYPEOF_VECTOR v, const uint8 c)
+{
+ bool result;
+#ifdef USE_SSE2
+ __m128i sub;
+#endif
+
+#if !defined(USE_SSE2) || defined(USE_ASSERT_CHECKING)
+ const char* s = (const char*) &v;
+#endif
+
+ /* pre-compute the result for assert checking */
+#ifdef USE_ASSERT_CHECKING
+ bool assert_result = false;
+
+ for (int j = 0; j < SIZEOF_VECTOR; j++)
+ {
+ if (s[j] <= c)
+ {
+ assert_result = true;
+ break;
+ }
+ }
+#endif /* USE_ASSERT_CHECKING */
+
+#ifdef USE_SSE2
+ /* use saturating subtraction to find bytes <= c, which will present as NUL bytes in 'sub' */
+ sub = _mm_subs_epu8(v, vector_broadcast(c));
+ return _mm_movemask_epi8(_mm_cmpeq_epi8(sub, _mm_setzero_si128()));
+#else
+ /* to find bytes <= c, we can use bitwise operations to find bytes < c+1, but it only works if c+1 <= 128 */
+ /* from https://graphics.stanford.edu/~seander/bithacks.html#HasLessInWord */
+ if (c + 1 <= 128)
+ return (v - vector_broadcast(c + 1)) & ~v & vector_broadcast(0x80);
+ else
+ {
+ /* one byte at a time */
+ for (int j = 0; j < SIZEOF_VECTOR; j++)
+ {
+ if (s[j] <= c)
+ {
+ result = true;
+ break;
+ }
+ }
+ }
+#endif
+
+#ifdef USE_ASSERT_CHECKING
+ Assert(assert_result == result);
+#endif /* USE_ASSERT_CHECKING */
+
+ return result;
+}
#endif /* SIMD_H */
diff --git a/src/test/modules/test_lfind/expected/test_lfind.out b/src/test/modules/test_lfind/expected/test_lfind.out
index 222c8fd7ff..1d4b14e703 100644
--- a/src/test/modules/test_lfind/expected/test_lfind.out
+++ b/src/test/modules/test_lfind/expected/test_lfind.out
@@ -4,9 +4,21 @@ CREATE EXTENSION test_lfind;
-- the operations complete without crashing or hanging and that none of their
-- internal sanity tests fail.
--
-SELECT test_lfind();
- test_lfind
-------------
+SELECT test_lfind8();
+ test_lfind8
+-------------
+
+(1 row)
+
+SELECT test_lfind8_le();
+ test_lfind8_le
+----------------
+
+(1 row)
+
+SELECT test_lfind32();
+ test_lfind32
+--------------
(1 row)
diff --git a/src/test/modules/test_lfind/sql/test_lfind.sql b/src/test/modules/test_lfind/sql/test_lfind.sql
index 899f1dd49b..766c640831 100644
--- a/src/test/modules/test_lfind/sql/test_lfind.sql
+++ b/src/test/modules/test_lfind/sql/test_lfind.sql
@@ -5,4 +5,6 @@ CREATE EXTENSION test_lfind;
-- the operations complete without crashing or hanging and that none of their
-- internal sanity tests fail.
--
-SELECT test_lfind();
+SELECT test_lfind8();
+SELECT test_lfind8_le();
+SELECT test_lfind32();
diff --git a/src/test/modules/test_lfind/test_lfind--1.0.sql b/src/test/modules/test_lfind/test_lfind--1.0.sql
index d82ab0567e..81801926ae 100644
--- a/src/test/modules/test_lfind/test_lfind--1.0.sql
+++ b/src/test/modules/test_lfind/test_lfind--1.0.sql
@@ -3,6 +3,14 @@
-- complain if script is sourced in psql, rather than via CREATE EXTENSION
\echo Use "CREATE EXTENSION test_lfind" to load this file. \quit
-CREATE FUNCTION test_lfind()
+CREATE FUNCTION test_lfind32()
+ RETURNS pg_catalog.void
+ AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION test_lfind8()
+ RETURNS pg_catalog.void
+ AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION test_lfind8_le()
RETURNS pg_catalog.void
AS 'MODULE_PATHNAME' LANGUAGE C;
diff --git a/src/test/modules/test_lfind/test_lfind.c b/src/test/modules/test_lfind/test_lfind.c
index a000746fb8..b853ddb609 100644
--- a/src/test/modules/test_lfind/test_lfind.c
+++ b/src/test/modules/test_lfind/test_lfind.c
@@ -18,10 +18,104 @@
PG_MODULE_MAGIC;
-PG_FUNCTION_INFO_V1(test_lfind);
+PG_FUNCTION_INFO_V1(test_lfind8);
+Datum
+test_lfind8(PG_FUNCTION_ARGS)
+{
+ unsigned char* str1 = (unsigned char*) "1234567890abcdef";
+ unsigned char* str2 = (unsigned char*) "1234567890abcdefZ";
+ size_t len1 = strlen((char*) str1);
+ size_t len2 = strlen((char*) str2);
+ uint8 key;
+
+ /* whole length of 16*/
+ Assert(len1 == 16);
+
+ key = 'X';
+ if (pg_lfind8(key, str1, len1))
+ elog(ERROR, "pg_lfind8() found nonexistent element '%c'", key);
+ key = 'f';
+ if (!pg_lfind8(key, str1, len1))
+ elog(ERROR, "pg_lfind8() did not find existing element '%c'", key);
+ /* include terminator */
+ key = '\0';
+ if (!pg_lfind8(key, str1, len1 + 1))
+ elog(ERROR, "pg_lfind8() did not find existing element '%c'", key);
+
+ /* restricted length */
+ key = '5';
+ if (pg_lfind8(key, str1, 4))
+ elog(ERROR, "pg_lfind8() found nonexistent element '%c'", key);
+ key = '4';
+ if (!pg_lfind8(key, str1, 4))
+ elog(ERROR, "pg_lfind8() did not find existing element '%c'", key);
+
+ /* test byte-wise path with string larger than vector size */
+ Assert(len2 > 16);
+
+ key = 'Y';
+ if (pg_lfind8(key, str2, len2))
+ elog(ERROR, "pg_lfind8() found nonexistent element '%c'", key);
+ key = 'Z';
+ if (!pg_lfind8(key, str2, len2))
+ elog(ERROR, "pg_lfind8() did not find existing element '%c'", key);
+
+ PG_RETURN_VOID();
+}
+
+
+PG_FUNCTION_INFO_V1(test_lfind8_le);
+Datum
+test_lfind8_le(PG_FUNCTION_ARGS)
+{
+ unsigned char* str1 = (unsigned char*) "A_CDEFGHIJKLMNO_";
+ unsigned char* str2 = (unsigned char*) "A_CDEFGHIJKLMNO_3";
+ size_t len1 = strlen((char*) str1);
+ size_t len2 = strlen((char*) str2);
+ uint8 key;
+
+ /* whole length of 16*/
+ Assert(len1 == 16);
+
+ /* search for char with value one less than minimum */
+ key = '@';
+ if (pg_lfind8_le(key, str1, len1))
+ elog(ERROR, "pg_lfind8_le() found nonexistent element <= '%c'", key);
+ /* search for minimum char */
+ key = 'A';
+ if (!pg_lfind8_le(key, str1, len1))
+ elog(ERROR, "pg_lfind8_le() did not find existing element <= '%c'", key);
+ key = 'B';
+ if (!pg_lfind8_le(key, str1, len1))
+ elog(ERROR, "pg_lfind8_le() did not find existing element <= '%c'", key);
+
+ /* search for terminating null by <= 0x01 */
+ key = 0x01;
+ if (!pg_lfind8_le(key, str1, len1 + 1))
+ elog(ERROR, "pg_lfind8_le() did not find existing element <= '%c'", key);
+
+ /* test byte-wise path with string larger than vector size */
+ Assert(len2 > 16);
+
+ /* search for char with value one less than minimum */
+ key = '2';
+ if (pg_lfind8_le(key, str2, len2))
+ elog(ERROR, "pg_lfind8() found nonexistent element <= '%c'", key);
+
+ /* search for minimum char */
+ key = '3';
+ if (!pg_lfind8_le(key, str2, len2))
+ elog(ERROR, "pg_lfind8() did not find existing element <= '%c'", key);
+ key = '4';
+ if (!pg_lfind8_le(key, str2, len2))
+ elog(ERROR, "pg_lfind8() did not find existing element <= '%c'", key);
+
+ PG_RETURN_VOID();
+}
+PG_FUNCTION_INFO_V1(test_lfind32);
Datum
-test_lfind(PG_FUNCTION_ARGS)
+test_lfind32(PG_FUNCTION_ARGS)
{
#define TEST_ARRAY_SIZE 135
uint32 test_array[TEST_ARRAY_SIZE] = {0};
