On Tue, 25 Jul 2023 at 17:34, Andres Freund <and...@anarazel.de> wrote:
> prep:
> COPY (SELECT generate_series(1, 2000000) a, (random() * 100000 - 50000)::int
> b, 3243423 c) TO '/tmp/lotsaints.copy';
> DROP TABLE lotsaints; CREATE UNLOGGED TABLE lotsaints(a int, b int, c int);
>
> benchmark:
> psql -qX -c 'truncate lotsaints' && pgbench -n -P1 -f <( echo "COPY lotsaints
> FROM '/tmp/lotsaints.copy';") -t 15
>
> I disabled turbo mode, pinned the server to a single core of my Xeon Gold
> 5215:
>
> HEAD: 812.690
>
> your patch: 821.354
>
> strtoint from 8692f6644e7: 824.543
>
> strtoint from 6b423ec677d^: 806.678
I'm surprised to see the imul version is faster. It's certainly not
what we found when working on 6b423ec67.
I've been fooling around a bit to try to learn what might be going on.
I wrote 2 patches:
1) pg_strtoint_imul.patch: Effectively reverts 6b423ec67 and puts the
code how it likely would have looked had I not done that work at all.
(I've assumed that the hex, octal, binary parsing would have been
added using the overflow multiplication functions just as the base 10
parsing).
2) pg_strtoint_optimize.patch: I've made another pass over the
functions with the current overflow checks and optimized the digit
checking code so that it can be done in a single check like: if (digit
< 10). This can be done by assigning the result of *ptr - '0' to an
unsigned char. Anything less than '0' will wrap around and anything
above '9' will remain so. I've also removed a few inefficiencies with
the isspace checking. We didn't need to do "while (*ptr &&
isspace(*ptr)). It's fine just to do while (isspace(*ptr)) since '\0'
isn't a space char. I also got rid of the isxdigit call. The
hexlookup array contains -1 for non-hex chars. We may as well check
the digit value is >= 0.
Here are the results I get using your test as quoted above:
master @ 62e9af4c63f + fix_COPY_DEFAULT.patch
latency average = 568.102 ms
master @ 62e9af4c63f + fix_COPY_DEFAULT.patch + pg_strtoint_optimize.patch
latency average = 531.238 ms
master @ 62e9af4c63f + fix_COPY_DEFAULT.patch + pg_strtoint_imul.patch
latency average = 559.333 ms (surprisingly also faster on my machine)
The optimized version of the pg_strtoint functions wins over the imul
patch. Could you test to see if this is also the case in your Xeon
machine?
> (when I say strtoint from, I did not replace the goto labels, so that part is
> unchanged and unrelated)
I didn't quite follow this.
I've not really studied the fix_COPY_DEFAULT.patch patch. Is there a
reason to delay committing that? It would be good to eliminate that
as a variable for the current performance regression.
David
diff --git a/src/backend/utils/adt/numutils.c b/src/backend/utils/adt/numutils.c
index 471fbb7ee6..a6e9d00727 100644
--- a/src/backend/utils/adt/numutils.c
+++ b/src/backend/utils/adt/numutils.c
@@ -104,10 +104,10 @@ static const int8 hexlookup[128] = {
* pg_strtoint16() will throw ereport() upon bad input format or overflow;
* while pg_strtoint16_safe() instead returns such complaints in *escontext,
* if it's an ErrorSaveContext.
-*
- * NB: Accumulate input as an unsigned number, to deal with two's complement
+ *
+ * NB: Accumulate input as a negative number, to deal with two's complement
* representation of the most negative number, which can't be represented as a
- * signed positive number.
+ * positive number.
*/
int16
pg_strtoint16(const char *s)
@@ -120,7 +120,7 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
const char *ptr = s;
const char *firstdigit;
- uint16 tmp = 0;
+ int16 tmp = 0;
bool neg = false;
/* skip leading spaces */
@@ -145,10 +145,11 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
if (isxdigit((unsigned char) *ptr))
{
- if (unlikely(tmp > -(PG_INT16_MIN / 16)))
- goto out_of_range;
+ int8 digit = hexlookup[(unsigned char) *ptr++];
- tmp = tmp * 16 + hexlookup[(unsigned char)
*ptr++];
+ if (unlikely(pg_mul_s16_overflow(tmp, 16,
&tmp)) ||
+ unlikely(pg_sub_s16_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -169,10 +170,11 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
if (*ptr >= '0' && *ptr <= '7')
{
- if (unlikely(tmp > -(PG_INT16_MIN / 8)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 8 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s16_overflow(tmp, 8, &tmp))
||
+ unlikely(pg_sub_s16_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -193,10 +195,11 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
if (*ptr >= '0' && *ptr <= '1')
{
- if (unlikely(tmp > -(PG_INT16_MIN / 2)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 2 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s16_overflow(tmp, 2, &tmp))
||
+ unlikely(pg_sub_s16_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -217,10 +220,11 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
if (isdigit((unsigned char) *ptr))
{
- if (unlikely(tmp > -(PG_INT16_MIN / 10)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 10 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s16_overflow(tmp, 10,
&tmp)) ||
+ unlikely(pg_sub_s16_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -248,18 +252,15 @@ pg_strtoint16_safe(const char *s, Node *escontext)
if (unlikely(*ptr != '\0'))
goto invalid_syntax;
- if (neg)
+ if (!neg)
{
- /* check the negative equivalent will fit without overflowing */
- if (tmp > (uint16) (-(PG_INT16_MIN + 1)) + 1)
+ /* could fail if input is most negative number */
+ if (unlikely(tmp == PG_INT16_MIN))
goto out_of_range;
- return -((int16) tmp);
+ tmp = -tmp;
}
- if (tmp > PG_INT16_MAX)
- goto out_of_range;
-
- return (int16) tmp;
+ return tmp;
out_of_range:
ereturn(escontext, 0,
@@ -298,7 +299,7 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
const char *ptr = s;
const char *firstdigit;
- uint32 tmp = 0;
+ int32 tmp = 0;
bool neg = false;
/* skip leading spaces */
@@ -323,10 +324,11 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
if (isxdigit((unsigned char) *ptr))
{
- if (unlikely(tmp > -(PG_INT32_MIN / 16)))
- goto out_of_range;
+ int8 digit = hexlookup[(unsigned char) *ptr++];
- tmp = tmp * 16 + hexlookup[(unsigned char)
*ptr++];
+ if (unlikely(pg_mul_s32_overflow(tmp, 16,
&tmp)) ||
+ unlikely(pg_sub_s32_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -347,10 +349,11 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
if (*ptr >= '0' && *ptr <= '7')
{
- if (unlikely(tmp > -(PG_INT32_MIN / 8)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 8 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s32_overflow(tmp, 8, &tmp))
||
+ unlikely(pg_sub_s32_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -371,10 +374,11 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
if (*ptr >= '0' && *ptr <= '1')
{
- if (unlikely(tmp > -(PG_INT32_MIN / 2)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 2 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s32_overflow(tmp, 2, &tmp))
||
+ unlikely(pg_sub_s32_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -395,10 +399,11 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
if (isdigit((unsigned char) *ptr))
{
- if (unlikely(tmp > -(PG_INT32_MIN / 10)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 10 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s32_overflow(tmp, 10,
&tmp)) ||
+ unlikely(pg_sub_s32_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -426,18 +431,15 @@ pg_strtoint32_safe(const char *s, Node *escontext)
if (unlikely(*ptr != '\0'))
goto invalid_syntax;
- if (neg)
+ if (!neg)
{
- /* check the negative equivalent will fit without overflowing */
- if (tmp > (uint32) (-(PG_INT32_MIN + 1)) + 1)
+ /* could fail if input is most negative number */
+ if (unlikely(tmp == PG_INT32_MIN))
goto out_of_range;
- return -((int32) tmp);
+ tmp = -tmp;
}
- if (tmp > PG_INT32_MAX)
- goto out_of_range;
-
- return (int32) tmp;
+ return tmp;
out_of_range:
ereturn(escontext, 0,
@@ -461,9 +463,9 @@ invalid_syntax:
* while pg_strtoint64_safe() instead returns such complaints in *escontext,
* if it's an ErrorSaveContext.
*
- * NB: Accumulate input as an unsigned number, to deal with two's complement
+ * NB: Accumulate input as a negative number, to deal with two's complement
* representation of the most negative number, which can't be represented as a
- * signed positive number.
+ * positive number.
*/
int64
pg_strtoint64(const char *s)
@@ -476,7 +478,7 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
const char *ptr = s;
const char *firstdigit;
- uint64 tmp = 0;
+ int64 tmp = 0;
bool neg = false;
/* skip leading spaces */
@@ -501,10 +503,11 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
if (isxdigit((unsigned char) *ptr))
{
- if (unlikely(tmp > -(PG_INT64_MIN / 16)))
- goto out_of_range;
+ int8 digit = hexlookup[(unsigned char) *ptr++];
- tmp = tmp * 16 + hexlookup[(unsigned char)
*ptr++];
+ if (unlikely(pg_mul_s64_overflow(tmp, 16,
&tmp)) ||
+ unlikely(pg_sub_s64_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -525,10 +528,11 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
if (*ptr >= '0' && *ptr <= '7')
{
- if (unlikely(tmp > -(PG_INT64_MIN / 8)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 8 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s64_overflow(tmp, 8, &tmp))
||
+ unlikely(pg_sub_s64_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -549,10 +553,11 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
if (*ptr >= '0' && *ptr <= '1')
{
- if (unlikely(tmp > -(PG_INT64_MIN / 2)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 2 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s64_overflow(tmp, 2, &tmp))
||
+ unlikely(pg_sub_s64_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -573,10 +578,11 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
if (isdigit((unsigned char) *ptr))
{
- if (unlikely(tmp > -(PG_INT64_MIN / 10)))
- goto out_of_range;
+ int8 digit = (*ptr++ - '0');
- tmp = tmp * 10 + (*ptr++ - '0');
+ if (unlikely(pg_mul_s64_overflow(tmp, 10,
&tmp)) ||
+ unlikely(pg_sub_s64_overflow(tmp,
digit, &tmp)))
+ goto out_of_range;
}
else if (*ptr == '_')
{
@@ -604,18 +610,15 @@ pg_strtoint64_safe(const char *s, Node *escontext)
if (unlikely(*ptr != '\0'))
goto invalid_syntax;
- if (neg)
+ if (!neg)
{
- /* check the negative equivalent will fit without overflowing */
- if (tmp > (uint64) (-(PG_INT64_MIN + 1)) + 1)
+ /* could fail if input is most negative number */
+ if (unlikely(tmp == PG_INT64_MIN))
goto out_of_range;
- return -((int64) tmp);
+ tmp = -tmp;
}
- if (tmp > PG_INT64_MAX)
- goto out_of_range;
-
- return (int64) tmp;
+ return tmp;
out_of_range:
ereturn(escontext, 0,
diff --git a/src/backend/utils/adt/numutils.c b/src/backend/utils/adt/numutils.c
index 471fbb7ee6..12d8b7eb69 100644
--- a/src/backend/utils/adt/numutils.c
+++ b/src/backend/utils/adt/numutils.c
@@ -124,7 +124,7 @@ pg_strtoint16_safe(const char *s, Node *escontext)
bool neg = false;
/* skip leading spaces */
- while (likely(*ptr) && isspace((unsigned char) *ptr))
+ while (isspace((unsigned char) *ptr))
ptr++;
/* handle sign */
@@ -141,14 +141,17 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (isxdigit((unsigned char) *ptr))
+ int8 digit = hexlookup[(unsigned char) *ptr];
+
+ if (likely(digit >= 0))
{
if (unlikely(tmp > -(PG_INT16_MIN / 16)))
goto out_of_range;
- tmp = tmp * 16 + hexlookup[(unsigned char)
*ptr++];
+ tmp = tmp * 16 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -165,14 +168,17 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (*ptr >= '0' && *ptr <= '7')
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 8))
{
if (unlikely(tmp > -(PG_INT16_MIN / 8)))
goto out_of_range;
- tmp = tmp * 8 + (*ptr++ - '0');
+ tmp = tmp * 8 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -189,14 +195,17 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (*ptr >= '0' && *ptr <= '1')
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 2))
{
if (unlikely(tmp > -(PG_INT16_MIN / 2)))
goto out_of_range;
- tmp = tmp * 2 + (*ptr++ - '0');
+ tmp = tmp * 2 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -213,14 +222,17 @@ pg_strtoint16_safe(const char *s, Node *escontext)
{
firstdigit = ptr;
- while (*ptr)
+ while (1)
{
- if (isdigit((unsigned char) *ptr))
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 10))
{
if (unlikely(tmp > -(PG_INT16_MIN / 10)))
goto out_of_range;
- tmp = tmp * 10 + (*ptr++ - '0');
+ tmp = tmp * 10 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -302,7 +314,7 @@ pg_strtoint32_safe(const char *s, Node *escontext)
bool neg = false;
/* skip leading spaces */
- while (likely(*ptr) && isspace((unsigned char) *ptr))
+ while (isspace((unsigned char) *ptr))
ptr++;
/* handle sign */
@@ -319,14 +331,17 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (isxdigit((unsigned char) *ptr))
+ int8 digit = hexlookup[(unsigned char) *ptr];
+
+ if (likely(digit >= 0))
{
if (unlikely(tmp > -(PG_INT32_MIN / 16)))
goto out_of_range;
- tmp = tmp * 16 + hexlookup[(unsigned char)
*ptr++];
+ tmp = tmp * 16 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -343,14 +358,17 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (*ptr >= '0' && *ptr <= '7')
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 8))
{
if (unlikely(tmp > -(PG_INT32_MIN / 8)))
goto out_of_range;
- tmp = tmp * 8 + (*ptr++ - '0');
+ tmp = tmp * 8 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -367,14 +385,17 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (*ptr >= '0' && *ptr <= '1')
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 2))
{
if (unlikely(tmp > -(PG_INT32_MIN / 2)))
goto out_of_range;
- tmp = tmp * 2 + (*ptr++ - '0');
+ tmp = tmp * 2 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -391,14 +412,17 @@ pg_strtoint32_safe(const char *s, Node *escontext)
{
firstdigit = ptr;
- while (*ptr)
+ while (1)
{
- if (isdigit((unsigned char) *ptr))
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 10))
{
if (unlikely(tmp > -(PG_INT32_MIN / 10)))
goto out_of_range;
- tmp = tmp * 10 + (*ptr++ - '0');
+ tmp = tmp * 10 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -420,7 +444,7 @@ pg_strtoint32_safe(const char *s, Node *escontext)
goto invalid_syntax;
/* allow trailing whitespace, but not other trailing chars */
- while (*ptr != '\0' && isspace((unsigned char) *ptr))
+ while (isspace((unsigned char) *ptr))
ptr++;
if (unlikely(*ptr != '\0'))
@@ -480,7 +504,7 @@ pg_strtoint64_safe(const char *s, Node *escontext)
bool neg = false;
/* skip leading spaces */
- while (*ptr && isspace((unsigned char) *ptr))
+ while (isspace((unsigned char) *ptr))
ptr++;
/* handle sign */
@@ -497,14 +521,17 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (isxdigit((unsigned char) *ptr))
+ int8 digit = hexlookup[(unsigned char) *ptr];
+
+ if (likely(digit >= 0))
{
if (unlikely(tmp > -(PG_INT64_MIN / 16)))
goto out_of_range;
- tmp = tmp * 16 + hexlookup[(unsigned char)
*ptr++];
+ tmp = tmp * 16 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -521,14 +548,17 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (*ptr >= '0' && *ptr <= '7')
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 8))
{
if (unlikely(tmp > -(PG_INT64_MIN / 8)))
goto out_of_range;
- tmp = tmp * 8 + (*ptr++ - '0');
+ tmp = tmp * 8 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -545,14 +575,17 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
firstdigit = ptr += 2;
- while (*ptr)
+ while (1)
{
- if (*ptr >= '0' && *ptr <= '1')
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 2))
{
if (unlikely(tmp > -(PG_INT64_MIN / 2)))
goto out_of_range;
- tmp = tmp * 2 + (*ptr++ - '0');
+ tmp = tmp * 2 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -569,14 +602,17 @@ pg_strtoint64_safe(const char *s, Node *escontext)
{
firstdigit = ptr;
- while (*ptr)
+ while (1)
{
- if (isdigit((unsigned char) *ptr))
+ unsigned char digit = *ptr - '0';
+
+ if (likely(digit < 10))
{
if (unlikely(tmp > -(PG_INT64_MIN / 10)))
goto out_of_range;
- tmp = tmp * 10 + (*ptr++ - '0');
+ tmp = tmp * 10 + digit;
+ ptr++;
}
else if (*ptr == '_')
{
@@ -598,7 +634,7 @@ pg_strtoint64_safe(const char *s, Node *escontext)
goto invalid_syntax;
/* allow trailing whitespace, but not other trailing chars */
- while (*ptr != '\0' && isspace((unsigned char) *ptr))
+ while (isspace((unsigned char) *ptr))
ptr++;
if (unlikely(*ptr != '\0'))
