perf: check whole UTF-8 strings without waiting for each block's result

valid_utf8_string() checks long strings 16 bytes at a time in
vector_utf8_valid_blocks(), which left its loop at the first block with
an error. Clang compiles "if (error) break; block += 16;" into a
conditional select on the block pointer, so each block waited for the
complete check of the block before: 28 cycles for 16 bytes on Apple M1.

The errors of all blocks are now collected and tested once at the end.
ASCII blocks are skipped in a loop of their own that needs none of the
lookup tables, and from the first block that is not ASCII, eight blocks
are checked without testing for ASCII in between, which was a
mispredicted branch for text with a few non-ASCII characters. If any
block had an error, the caller finds it one sequence at a time from the
start of the string, as before from the block with the error.

The check is used by the binary writers and readers with an error
handler other than keep, and always by BON8. On Apple M1 (Apple Clang),
it needs 0.27 instead of 1.78 cycles per byte for non-ASCII text, and
to_msgpack() with error_handler_t::strict is 24 to 43 % faster for such
text. With GCC 15 on x86-64, where the loop was not affected, the check
itself is 13 to 35 % faster.

A new test puts well-formed and ill-formed sequences at every offset of
strings that are longer than eight blocks.

Signed-off-by: Niels Lohmann <mail@nlohmann.me>

include/nlohmann/detail/input/simd_scan.hpp

@@ -324,15 +324,23 @@ JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline bool vector_utf8_run(cons
 }
 
 /*!
-@brief skip 16-byte blocks of well-formed UTF-8 at p (a character boundary)
+@brief vector_utf8_valid_blocks() from the first block that is not ASCII
 
-Like vector_utf8_run, but quotes, backslashes, and control characters are
+@a block is that block, and @a first is where the check began (only ASCII up
-ordinary characters. Returns the start of the first block in which the check
+to @a block). Returns where fewer than 16 bytes are left, or @a first if the
-found an error (which may belong to a sequence begun in the block before) or
+check found an error in any block.
-where fewer than 16 bytes are left; a sequence may be cut at that point, so the
+
-caller resumes at its start, one sequence at a time.
+From a block that is not ASCII on, eight blocks are checked without testing
+for ASCII again: in text with a few non-ASCII characters (accented letters),
+every other block has one, and the test would be a mispredicted branch for
+most blocks. After them, ASCII blocks are skipped with one test again.
+
+The errors of all blocks are collected and tested once at the end. A test
+after each block lets the compiler advance the position with a conditional
+select on its result (Apple Clang 17 does), and each block then waits for the
+check of the block before: 28 cycles for 16 bytes instead of 4 on Apple M1.
 */
-JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vector_utf8_valid_blocks(const unsigned char* block, const unsigned char* e) noexcept
+JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vector_utf8_check_blocks(const unsigned char* first, const unsigned char* block, const unsigned char* e) noexcept
 {
     using lookup = utf8_lookup4<>;
 #if JSON_SCAN_NEON
@@ -340,27 +348,29 @@ JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vect
     const uint8x16_t t1l = vld1q_u8(lookup::byte_1_low.data());
     const uint8x16_t t2h = vld1q_u8(lookup::byte_2_high.data());
     uint8x16_t prev = vdupq_n_u8(0);
+    uint8x16_t error = vdupq_n_u8(0);
     while (e - block >= 16)
     {
-        const uint8x16_t in = vld1q_u8(block);
+        const uint8x16_t ahead = vld1q_u8(block);
-        if (vmaxvq_u8(vorrq_u8(in, prev)) < 0x80)
+        if (vmaxvq_u8(vorrq_u8(ahead, prev)) < 0x80)
         {
             // ASCII here, and no sequence from the block before to finish
-            prev = in;
+            prev = ahead;
             block += 16;
             continue;
         }
-        const uint8x16_t prev1 = vextq_u8(prev, in, 15);
+        for (int blocks = 8; blocks != 0 && e - block >= 16; --blocks)
-        const uint8x16_t sc = vandq_u8(vandq_u8(vqtbl1q_u8(t1h, vshrq_n_u8(prev1, 4)), vqtbl1q_u8(t1l, vandq_u8(prev1, vdupq_n_u8(0x0F)))), vqtbl1q_u8(t2h, vshrq_n_u8(in, 4)));
-        const uint8x16_t must23 = vorrq_u8(vqsubq_u8(vextq_u8(prev, in, 14), vdupq_n_u8(0xE0 - 0x80)), vqsubq_u8(vextq_u8(prev, in, 13), vdupq_n_u8(0xF0 - 0x80)));
-        const uint8x16_t err = veorq_u8(vandq_u8(must23, vdupq_n_u8(0x80)), sc);
-        if (vmaxvq_u8(err) != 0)
         {
-            break;
+            const uint8x16_t in = vld1q_u8(block);
+            const uint8x16_t prev1 = vextq_u8(prev, in, 15);
+            const uint8x16_t sc = vandq_u8(vandq_u8(vqtbl1q_u8(t1h, vshrq_n_u8(prev1, 4)), vqtbl1q_u8(t1l, vandq_u8(prev1, vdupq_n_u8(0x0F)))), vqtbl1q_u8(t2h, vshrq_n_u8(in, 4)));
+            const uint8x16_t must23 = vorrq_u8(vqsubq_u8(vextq_u8(prev, in, 14), vdupq_n_u8(0xE0 - 0x80)), vqsubq_u8(vextq_u8(prev, in, 13), vdupq_n_u8(0xF0 - 0x80)));
+            error = vorrq_u8(error, veorq_u8(vandq_u8(must23, vdupq_n_u8(0x80)), sc));
+            prev = in;
+            block += 16;
         }
-        prev = in;
-        block += 16;
     }
+    return vmaxvq_u8(error) != 0 ? first : block;
 #else
     const __m128i t1h = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(lookup::byte_1_high.data()))); // NOLINT(bugprone-casting-through-void)
     const __m128i t1l = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(lookup::byte_1_low.data()))); // NOLINT(bugprone-casting-through-void)
@@ -368,30 +378,62 @@ JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vect
     const __m128i nibble = _mm_set1_epi8(0x0F);
     const __m128i zero = _mm_setzero_si128();
     __m128i prev = zero;
+    __m128i error = zero;
     while (e - block >= 16)
     {
-        const __m128i in = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(block))); // NOLINT(bugprone-casting-through-void)
+        const __m128i ahead = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(block))); // NOLINT(bugprone-casting-through-void)
-        if (_mm_movemask_epi8(_mm_or_si128(in, prev)) == 0)
+        if (_mm_movemask_epi8(_mm_or_si128(ahead, prev)) == 0)
         {
-            prev = in;
+            prev = ahead;
             block += 16;
             continue;
         }
-        const __m128i prev1 = _mm_alignr_epi8(in, prev, 15);
+        for (int blocks = 8; blocks != 0 && e - block >= 16; --blocks)
-        const __m128i sc = _mm_and_si128(_mm_and_si128(_mm_shuffle_epi8(t1h, _mm_and_si128(_mm_srli_epi16(prev1, 4), nibble)),
-                                         _mm_shuffle_epi8(t1l, _mm_and_si128(prev1, nibble))),
-                                         _mm_shuffle_epi8(t2h, _mm_and_si128(_mm_srli_epi16(in, 4), nibble)));
-        const __m128i must23 = _mm_or_si128(_mm_subs_epu8(_mm_alignr_epi8(in, prev, 14), _mm_set1_epi8(0xE0 - 0x80)),
-                                            _mm_subs_epu8(_mm_alignr_epi8(in, prev, 13), _mm_set1_epi8(0xF0 - 0x80)));
-        const __m128i err = _mm_xor_si128(_mm_and_si128(must23, _mm_set1_epi8(static_cast<char>(-128))), sc);
-        if (_mm_movemask_epi8(_mm_cmpeq_epi8(err, zero)) != 0xFFFF)
         {
-            break;
+            const __m128i in = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(block))); // NOLINT(bugprone-casting-through-void)
+            const __m128i prev1 = _mm_alignr_epi8(in, prev, 15);
+            const __m128i sc = _mm_and_si128(_mm_and_si128(_mm_shuffle_epi8(t1h, _mm_and_si128(_mm_srli_epi16(prev1, 4), nibble)),
+                                             _mm_shuffle_epi8(t1l, _mm_and_si128(prev1, nibble))),
+                                             _mm_shuffle_epi8(t2h, _mm_and_si128(_mm_srli_epi16(in, 4), nibble)));
+            const __m128i must23 = _mm_or_si128(_mm_subs_epu8(_mm_alignr_epi8(in, prev, 14), _mm_set1_epi8(0xE0 - 0x80)),
+                                                _mm_subs_epu8(_mm_alignr_epi8(in, prev, 13), _mm_set1_epi8(0xF0 - 0x80)));
+            error = _mm_or_si128(error, _mm_xor_si128(_mm_and_si128(must23, _mm_set1_epi8(static_cast<char>(-128))), sc));
+            prev = in;
+            block += 16;
+        }
+    }
+    return _mm_movemask_epi8(_mm_cmpeq_epi8(error, zero)) != 0xFFFF ? first : block;
+#endif
+}
+
+/*!
+@brief skip 16-byte blocks of well-formed UTF-8 at p (a character boundary)
+
+Like vector_utf8_run, but quotes, backslashes, and control characters are
+ordinary characters. Returns where fewer than 16 bytes are left; a sequence may
+be cut at that point, so the caller resumes at its start, one sequence at a
+time. If the check found an error in any block, @a block itself is returned,
+for the caller to find the error one sequence at a time.
+
+ASCII blocks need one test each and none of the tables of the UTF-8 check,
+which starts at the first block that is not ASCII (see
+vector_utf8_check_blocks()). Out of line, like the check.
+*/
+JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vector_utf8_valid_blocks(const unsigned char* block, const unsigned char* e) noexcept
+{
+    const unsigned char* const first = block;
+    while (e - block >= 16)
+    {
+#if JSON_SCAN_NEON
+        if (JSON_HEDLEY_UNLIKELY(vmaxvq_u8(vld1q_u8(block)) >= 0x80))
+#else
+        if (JSON_HEDLEY_UNLIKELY(_mm_movemask_epi8(_mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(block)))) != 0)) // NOLINT(bugprone-casting-through-void)
+#endif
+        {
+            return vector_utf8_check_blocks(first, block, e);
         }
-        prev = in;
         block += 16;
     }
-#endif
     return block;
 }
 #endif

single_include/nlohmann/json.hpp

@@ -12593,15 +12593,23 @@ JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline bool vector_utf8_run(cons
 }
 
 /*!
-@brief skip 16-byte blocks of well-formed UTF-8 at p (a character boundary)
+@brief vector_utf8_valid_blocks() from the first block that is not ASCII
 
-Like vector_utf8_run, but quotes, backslashes, and control characters are
+@a block is that block, and @a first is where the check began (only ASCII up
-ordinary characters. Returns the start of the first block in which the check
+to @a block). Returns where fewer than 16 bytes are left, or @a first if the
-found an error (which may belong to a sequence begun in the block before) or
+check found an error in any block.
-where fewer than 16 bytes are left; a sequence may be cut at that point, so the
+
-caller resumes at its start, one sequence at a time.
+From a block that is not ASCII on, eight blocks are checked without testing
+for ASCII again: in text with a few non-ASCII characters (accented letters),
+every other block has one, and the test would be a mispredicted branch for
+most blocks. After them, ASCII blocks are skipped with one test again.
+
+The errors of all blocks are collected and tested once at the end. A test
+after each block lets the compiler advance the position with a conditional
+select on its result (Apple Clang 17 does), and each block then waits for the
+check of the block before: 28 cycles for 16 bytes instead of 4 on Apple M1.
 */
-JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vector_utf8_valid_blocks(const unsigned char* block, const unsigned char* e) noexcept
+JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vector_utf8_check_blocks(const unsigned char* first, const unsigned char* block, const unsigned char* e) noexcept
 {
     using lookup = utf8_lookup4<>;
 #if JSON_SCAN_NEON
@@ -12609,27 +12617,29 @@ JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vect
     const uint8x16_t t1l = vld1q_u8(lookup::byte_1_low.data());
     const uint8x16_t t2h = vld1q_u8(lookup::byte_2_high.data());
     uint8x16_t prev = vdupq_n_u8(0);
+    uint8x16_t error = vdupq_n_u8(0);
     while (e - block >= 16)
     {
-        const uint8x16_t in = vld1q_u8(block);
+        const uint8x16_t ahead = vld1q_u8(block);
-        if (vmaxvq_u8(vorrq_u8(in, prev)) < 0x80)
+        if (vmaxvq_u8(vorrq_u8(ahead, prev)) < 0x80)
         {
             // ASCII here, and no sequence from the block before to finish
-            prev = in;
+            prev = ahead;
             block += 16;
             continue;
         }
-        const uint8x16_t prev1 = vextq_u8(prev, in, 15);
+        for (int blocks = 8; blocks != 0 && e - block >= 16; --blocks)
-        const uint8x16_t sc = vandq_u8(vandq_u8(vqtbl1q_u8(t1h, vshrq_n_u8(prev1, 4)), vqtbl1q_u8(t1l, vandq_u8(prev1, vdupq_n_u8(0x0F)))), vqtbl1q_u8(t2h, vshrq_n_u8(in, 4)));
-        const uint8x16_t must23 = vorrq_u8(vqsubq_u8(vextq_u8(prev, in, 14), vdupq_n_u8(0xE0 - 0x80)), vqsubq_u8(vextq_u8(prev, in, 13), vdupq_n_u8(0xF0 - 0x80)));
-        const uint8x16_t err = veorq_u8(vandq_u8(must23, vdupq_n_u8(0x80)), sc);
-        if (vmaxvq_u8(err) != 0)
         {
-            break;
+            const uint8x16_t in = vld1q_u8(block);
+            const uint8x16_t prev1 = vextq_u8(prev, in, 15);
+            const uint8x16_t sc = vandq_u8(vandq_u8(vqtbl1q_u8(t1h, vshrq_n_u8(prev1, 4)), vqtbl1q_u8(t1l, vandq_u8(prev1, vdupq_n_u8(0x0F)))), vqtbl1q_u8(t2h, vshrq_n_u8(in, 4)));
+            const uint8x16_t must23 = vorrq_u8(vqsubq_u8(vextq_u8(prev, in, 14), vdupq_n_u8(0xE0 - 0x80)), vqsubq_u8(vextq_u8(prev, in, 13), vdupq_n_u8(0xF0 - 0x80)));
+            error = vorrq_u8(error, veorq_u8(vandq_u8(must23, vdupq_n_u8(0x80)), sc));
+            prev = in;
+            block += 16;
         }
-        prev = in;
-        block += 16;
     }
+    return vmaxvq_u8(error) != 0 ? first : block;
 #else
     const __m128i t1h = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(lookup::byte_1_high.data()))); // NOLINT(bugprone-casting-through-void)
     const __m128i t1l = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(lookup::byte_1_low.data()))); // NOLINT(bugprone-casting-through-void)
@@ -12637,30 +12647,62 @@ JSON_SCAN_SSSE3_TARGET JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vect
     const __m128i nibble = _mm_set1_epi8(0x0F);
     const __m128i zero = _mm_setzero_si128();
     __m128i prev = zero;
+    __m128i error = zero;
     while (e - block >= 16)
     {
-        const __m128i in = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(block))); // NOLINT(bugprone-casting-through-void)
+        const __m128i ahead = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(block))); // NOLINT(bugprone-casting-through-void)
-        if (_mm_movemask_epi8(_mm_or_si128(in, prev)) == 0)
+        if (_mm_movemask_epi8(_mm_or_si128(ahead, prev)) == 0)
         {
-            prev = in;
+            prev = ahead;
             block += 16;
             continue;
         }
-        const __m128i prev1 = _mm_alignr_epi8(in, prev, 15);
+        for (int blocks = 8; blocks != 0 && e - block >= 16; --blocks)
-        const __m128i sc = _mm_and_si128(_mm_and_si128(_mm_shuffle_epi8(t1h, _mm_and_si128(_mm_srli_epi16(prev1, 4), nibble)),
-                                         _mm_shuffle_epi8(t1l, _mm_and_si128(prev1, nibble))),
-                                         _mm_shuffle_epi8(t2h, _mm_and_si128(_mm_srli_epi16(in, 4), nibble)));
-        const __m128i must23 = _mm_or_si128(_mm_subs_epu8(_mm_alignr_epi8(in, prev, 14), _mm_set1_epi8(0xE0 - 0x80)),
-                                            _mm_subs_epu8(_mm_alignr_epi8(in, prev, 13), _mm_set1_epi8(0xF0 - 0x80)));
-        const __m128i err = _mm_xor_si128(_mm_and_si128(must23, _mm_set1_epi8(static_cast<char>(-128))), sc);
-        if (_mm_movemask_epi8(_mm_cmpeq_epi8(err, zero)) != 0xFFFF)
         {
-            break;
+            const __m128i in = _mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(block))); // NOLINT(bugprone-casting-through-void)
+            const __m128i prev1 = _mm_alignr_epi8(in, prev, 15);
+            const __m128i sc = _mm_and_si128(_mm_and_si128(_mm_shuffle_epi8(t1h, _mm_and_si128(_mm_srli_epi16(prev1, 4), nibble)),
+                                             _mm_shuffle_epi8(t1l, _mm_and_si128(prev1, nibble))),
+                                             _mm_shuffle_epi8(t2h, _mm_and_si128(_mm_srli_epi16(in, 4), nibble)));
+            const __m128i must23 = _mm_or_si128(_mm_subs_epu8(_mm_alignr_epi8(in, prev, 14), _mm_set1_epi8(0xE0 - 0x80)),
+                                                _mm_subs_epu8(_mm_alignr_epi8(in, prev, 13), _mm_set1_epi8(0xF0 - 0x80)));
+            error = _mm_or_si128(error, _mm_xor_si128(_mm_and_si128(must23, _mm_set1_epi8(static_cast<char>(-128))), sc));
+            prev = in;
+            block += 16;
+        }
+    }
+    return _mm_movemask_epi8(_mm_cmpeq_epi8(error, zero)) != 0xFFFF ? first : block;
+#endif
+}
+
+/*!
+@brief skip 16-byte blocks of well-formed UTF-8 at p (a character boundary)
+
+Like vector_utf8_run, but quotes, backslashes, and control characters are
+ordinary characters. Returns where fewer than 16 bytes are left; a sequence may
+be cut at that point, so the caller resumes at its start, one sequence at a
+time. If the check found an error in any block, @a block itself is returned,
+for the caller to find the error one sequence at a time.
+
+ASCII blocks need one test each and none of the tables of the UTF-8 check,
+which starts at the first block that is not ASCII (see
+vector_utf8_check_blocks()). Out of line, like the check.
+*/
+JSON_HEDLEY_NEVER_INLINE inline const unsigned char* vector_utf8_valid_blocks(const unsigned char* block, const unsigned char* e) noexcept
+{
+    const unsigned char* const first = block;
+    while (e - block >= 16)
+    {
+#if JSON_SCAN_NEON
+        if (JSON_HEDLEY_UNLIKELY(vmaxvq_u8(vld1q_u8(block)) >= 0x80))
+#else
+        if (JSON_HEDLEY_UNLIKELY(_mm_movemask_epi8(_mm_loadu_si128(static_cast<const __m128i*>(static_cast<const void*>(block)))) != 0)) // NOLINT(bugprone-casting-through-void)
+#endif
+        {
+            return vector_utf8_check_blocks(first, block, e);
         }
-        prev = in;
         block += 16;
     }
-#endif
     return block;
 }
 #endif

tests/src/unit-unicode.cpp

@@ -1669,3 +1669,60 @@ TEST_CASE("UTF-8 check of whole strings")
     CHECK(valid > 20000);
     CHECK(valid < 80000);
 }
+
+TEST_CASE("UTF-8 check of long strings")
+{
+    // Where a vector check is available, valid_utf8_string() skips ASCII 16
+    // bytes at a time, checks eight blocks from the first one that is not
+    // ASCII without testing for ASCII in between, and collects the errors of
+    // all blocks until the end. A sequence of each length and each kind of
+    // ill-formed sequence is put at every offset of strings that are longer
+    // than eight blocks: before, inside, at the end of, and after these
+    // blocks, and across every block boundary.
+    const char* const pieces[] = // NOLINT(cppcoreguidelines-avoid-c-arrays,hicpp-avoid-c-arrays,modernize-avoid-c-arrays)
+    {
+        "\xC3\xA9", "\xE2\x82\xAC", "\xE0\xA0\x80", "\xEF\xBF\xBF", "\xF0\x9F\x98\x80", "\xF4\x8F\xBF\xBF",
+        "\x80", "\xC0\xAF", "\xC3", "\xE2\x82", "\xED\xA0\x80", "\xF0\x9F", "\xF0\x9F\x98", "\xF4\x90\x80\x80", "\xFF"
+    };
+    // the text around the piece: ASCII only, a sequence at the start (so that
+    // the check of eight blocks starts there), and a sequence in every block
+    const char* const fillers[] = {"", "\xC3\xA9", "\xE2\x82\xAC"}; // NOLINT(cppcoreguidelines-avoid-c-arrays,hicpp-avoid-c-arrays,modernize-avoid-c-arrays)
+    std::size_t valid = 0;
+    std::size_t invalid = 0;
+    for (const char* const piece : pieces)
+    {
+        for (std::size_t filler = 0; filler < 3; ++filler)
+        {
+            std::string base;
+            if (filler == 1)
+            {
+                base += fillers[1];
+            }
+            while (base.size() < 300)
+            {
+                base += (filler == 2 && base.size() % 16 == 5) ? std::string(fillers[2]) : std::string("x");
+            }
+            for (std::size_t offset = 0; offset <= base.size(); ++offset)
+            {
+                // (not inside a sequence of the text around the piece)
+                if (offset < base.size() && (static_cast<unsigned char>(base[offset]) & 0xC0u) == 0x80u)
+                {
+                    continue;
+                }
+                std::string s = base;
+                s.insert(offset, piece);
+                const auto* data = reinterpret_cast<const unsigned char*>(s.data());
+                const bool expected = nlohmann::detail::is_valid_utf8(s);
+                (expected ? valid : invalid) += 1;
+                CAPTURE(offset)
+                CAPTURE(filler)
+                CAPTURE(piece)
+                CHECK(nlohmann::detail::valid_utf8_string(data, s.size()) == expected);
+                CHECK((nlohmann::detail::valid_utf8_prefix(data, s.size()) == s.size()) == expected);
+            }
+        }
+    }
+    // the six well-formed pieces are valid at every offset, the others at none
+    CHECK(valid * 9 == invalid * 6);
+    CHECK(valid > 5000);
+}