Own the parallel stream driver

CMakeLists.txt

@@ -28,14 +28,20 @@ find_package(Threads REQUIRED)
 list(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_LIST_DIR}/cmake)
 include(get_cpm)
 
-# --- simdjson: the PR #2788 branch, pinned by commit -----------------------
-# parse_many_parallel lives in include/simdjson/ondemand_parallel.h on this
-# branch. Pinning by SHA keeps the paper's numbers reproducible even if the
-# branch is rebased or merged.
+# --- simdjson --------------------------------------------------------------
+# The parallel stream driver is ours (src/parallel_stream.h), so this is a plain
+# master commit rather than a patched branch. It is not a release: simdjson cuts
+# releases from a 4.6.x branch, and neither stream_format::newline_delimited nor
+# simdjson::slice_at (both merged to master in #2803) has shipped in one yet. A
+# release still builds -- the probe below detects their absence and the driver
+# falls back to its own slicing -- but it measures a different thing, so the
+# reported numbers come from master. Pinned by SHA to keep the paper
+# reproducible.
+
 CPMAddPackage(
   NAME simdjson
   GITHUB_REPOSITORY simdjson/simdjson
-  GIT_TAG 88186d59  # PR #2788
+  GIT_TAG 3839ac681a4a6b4fd09b4d7e03229b35bcd909d7  # master, 2026-08-26
   OPTIONS "SIMDJSON_DEVELOPER_MODE OFF" "BUILD_SHARED_LIBS OFF")
 
 # --- lemire/counters: hardware performance counters (header only) ----------
@@ -69,6 +75,29 @@ target_compile_options(pison PUBLIC -include cstdint)
 target_compile_features(pison PUBLIC cxx_std_11)
 set_target_properties(pison PROPERTIES POSITION_INDEPENDENT_CODE ON)
 
+# The parallel stream driver lives in this repository (src/parallel_stream.h),
+# not in simdjson. It goes faster when simdjson can skip the tail of a partially
+# read document, which needs stream_format::newline_delimited; detect it rather
+# than require it, so the benchmark still builds against released simdjson.
+set(NEWLINE_DELIMITED_POS -1)
+set(SLICE_AT_POS -1)
+if(EXISTS ${simdjson_SOURCE_DIR}/include/simdjson/base.h)
+  file(READ ${simdjson_SOURCE_DIR}/include/simdjson/base.h SIMDJSON_BASE_H)
+  string(FIND "${SIMDJSON_BASE_H}" "newline_delimited" NEWLINE_DELIMITED_POS)
+endif()
+if(EXISTS ${simdjson_SOURCE_DIR}/include/simdjson/padded_string_view.h)
+  file(READ ${simdjson_SOURCE_DIR}/include/simdjson/padded_string_view.h SIMDJSON_PSV_H)
+  string(FIND "${SIMDJSON_PSV_H}" "slice_at" SLICE_AT_POS)
+endif()
+if(NEWLINE_DELIMITED_POS GREATER -1 AND SLICE_AT_POS GREATER -1)
+  set(NEWLINE_DELIMITED 1)
+  message(STATUS "simdjson has stream_format::newline_delimited")
+else()
+  set(NEWLINE_DELIMITED 0)
+  message(STATUS "simdjson lacks newline_delimited/slice_at; the parallel "
+                 "engine falls back to its own slicing")
+endif()
+
 # --- conventional DOM parsers ----------------------------------------------
 # Baselines: what a practitioner uses today. Each is optional; a library that
 # does not resolve drops its engine from the run rather than breaking the build.
@@ -158,6 +187,7 @@ add_executable(jsonbench
   ${DOM_SOURCES})
 target_include_directories(jsonbench PRIVATE src)
 target_compile_definitions(jsonbench PRIVATE
+  JSONBENCH_HAVE_NEWLINE_DELIMITED=${NEWLINE_DELIMITED}
   JSONBENCH_HAVE_YYJSON=${HAVE_YYJSON}
   JSONBENCH_HAVE_RAPIDJSON=${HAVE_RAPIDJSON}
   JSONBENCH_HAVE_BOOST_JSON=${HAVE_BOOST_JSON}

README.md

@@ -1,9 +1,8 @@
 # jsonstreambench
 
 A fair benchmark for parsing **streams of JSON documents**, comparing
-[simdjson](https://github.com/simdjson/simdjson) — including the experimental
-parallel stream parser of
-[PR #2788](https://github.com/simdjson/simdjson/pull/2788) — against
+[simdjson](https://github.com/simdjson/simdjson) — driven across many threads by
+the slicing rule in `src/parallel_stream.h` — against
 [Pison](https://github.com/AutomataLab/Pison), on Pison's own corpus and
 queries.
 
@@ -27,18 +26,31 @@ jsonbench --dataset <file.ndjson> [options]
                        (default: inferred from the filename)
   --threads a,b,c      thread counts to sweep (default: 1..hw, doubling)
   --reps <n>           repetitions per configuration, best wins (default 3)
-  --slice-kb <n>       parse_many_parallel slice size (default 1024)
+  --slice-kb <n>       parallel slice size (default 64)
+  --assign <mode>      slice assignment: static (default) or dynamic
+  --batch-mb <n>       iterate_many batch size (default 16)
   --sections <list>    load,verify,single,scaling,e2e,format (default: all
                        but format)
   --single-record      treat the input as one bulky JSON document
   --verify             check that the engines agree, then exit
   --dump <n>           print the first n extracted values from each engine
+  --engine-only <name> run only this engine in the scaling section
+  --impl <name>        force a simdjson kernel (haswell, icelake, ...)
+  --levels <n>         override Pison's level_num
 ```
 
 The `format` section compares comma-delimited against newline-delimited
 encoding of the same records, serially and on simdjson's two-thread pipeline.
 
-Output is one `RESULT key=value ...` line per measured configuration.
+Output is one `RESULT key=value ...` line per measured configuration. Each
+carries `spread_pct`, the gap between the slowest and fastest repetition
+relative to the fastest: only the best repetition is reported, and how
+repeatable it was varies enormously between machines.
+
+`--engine-only` narrows the scaling section to one engine, which is what makes
+an aggregate profile interpretable — `perf stat -a` cannot attribute a counter
+to an engine when several run in the same process. Pair it with
+`--sections scaling` so nothing else runs either.
 
 
 ## How the comparison is kept fair
@@ -84,6 +96,48 @@ engines spawn workers internally, so a multi-threaded reading would be wrong
 rather than noisy. Parallel runs report wall-clock throughput plus CPU seconds
 per gigabyte from `getrusage`, which does aggregate all threads.
 
+## The parallel driver
+
+`src/parallel_stream.h` is this repository's own driver, not a simdjson API. An
+earlier version lived in simdjson as an experimental header
+([PR #2788](https://github.com/simdjson/simdjson/pull/2788)); it belongs in user
+code instead, because nothing in it needs to be inside the library — it is a
+slicing rule plus a thread pool over the public `iterate_many` interface — and
+keeping it out means a caller can adapt the decomposition to their own pipeline
+rather than accept ours. simdjson is therefore pinned to an ordinary master
+commit rather than a patched branch. `src/dom_parallel.h` applies the same rule
+to yyjson, RapidJSON, Boost.JSON and nlohmann: the slicing does not know what
+parses a document, and running the conventional parsers under it is how that
+claim is checked rather than asserted.
+
+The rule: cut the input into fixed-size slices, snap both ends forward to the
+next delimiter so slices abut and no document is split, and give each worker its
+own parser and its own output vector. Nothing is shared on the hot path, and
+values keep their order within a worker but not across the input. This needs a
+delimiter that cannot occur inside a JSON value — a line feed for NDJSON, a
+record separator (0x1E) for RFC 7464 — so comma-delimited input cannot be sliced
+this way at all, its top-level commas being findable only by a serial structural
+scan. Our corpus is strictly one document per line, so simdjson is told that too
+with `stream_format::newline_delimited`, which lets it skip the tail of a
+partially read document rather than walk it. That format and
+`simdjson::slice_at` are detected at configure time rather than required: every
+simdjson *release* lacks both, since releases are cut from a `4.6.x` branch, and
+against one the driver falls back to slicing with `memchr`.
+
+Two knobs control it, and the right values depend on the corpus rather than on
+the machine. `--assign` (default `static`) gives each worker one contiguous run
+of slices; `dynamic` instead has workers claim the next free slice from a shared
+counter, which scatters their regions across the input and, once slices are
+small enough for the counter to be contended, costs a large fraction of the
+throughput. `--slice-kb` (default 64) sets the slice. Both changed when this
+driver landed — they were 1024 KB and `dynamic` — so numbers collected before
+that are not comparable to numbers collected after it. Both were also tuned on
+the six Pison datasets, whose documents are small, and **they do not transfer to
+a corpus of bulky records**: a document larger than a slice is rescanned once
+per overlapping slice, which costs throughput in proportion to the square of the
+document over the slice. Results stay correct at every size; only throughput
+suffers. Raise `--slice-kb` above the longest document for such a corpus.
+
 ## Corpus
 
 `./datasets.sh` obtains the whole corpus:

src/dom_engine.h

@@ -47,6 +47,13 @@ extraction run_serial(library lib, const char *data, size_t size, query_id q,
 extraction run_parallel(library lib, const char *data, size_t size, query_id q,
                         size_t threads, size_t slice_bytes, size_t longest);
 
+// Mirrors parallel::options::static_partition for the DOM drivers, which share
+// run_sliced rather than a per-call options struct. Set before a run.
+inline bool &static_partition_flag() {
+  static bool value = true;
+  return value;
+}
+
 } // namespace dom
 } // namespace jsonbench
 

src/dom_parallel.h

@@ -12,6 +12,7 @@
 #define JSONBENCH_DOM_PARALLEL_H
 
 #include "common.h"
+#include "dom_engine.h"
 #include "dom_queries.h"
 
 #include <atomic>
@@ -41,16 +42,25 @@ extraction run_sliced(const char *data, size_t length, size_t threads,
   if (threads == 0) { threads = 1; }
   if (slice_bytes == 0) { slice_bytes = 256 * 1024; }
   std::atomic<size_t> cursor{0};
+  const bool stat = static_partition_flag();
+  const size_t slices = (length + slice_bytes - 1) / slice_bytes;
   std::vector<extraction> shards(threads);
   std::vector<std::thread> workers;
   workers.reserve(threads);
   for (size_t i = 0; i < threads; i++) {
     workers.emplace_back([&, i] {
       auto parse_slice = make_worker();
       extraction &mine = shards[i];
+      size_t next = slices * i / threads;
+      const size_t last = slices * (i + 1) / threads;
       for (;;) {
-        const size_t raw = cursor.fetch_add(slice_bytes,
-                                            std::memory_order_relaxed);
+        size_t raw;
+        if (stat) {
+          if (next >= last) { break; }
+          raw = next++ * slice_bytes;
+        } else {
+          raw = cursor.fetch_add(slice_bytes, std::memory_order_relaxed);
+        }
         if (raw >= length) { break; }
         const size_t begin = (raw == 0) ? 0 : snap_to_newline(data, length, raw);
         const size_t end = snap_to_newline(data, length, raw + slice_bytes);

src/main.cpp

@@ -1,5 +1,5 @@
-// jsonbench -- simdjson (incl. PR #2788 parse_many_parallel) vs Pison on a
-// stream of JSON documents.
+// jsonbench -- simdjson (driven in parallel by src/parallel_stream.h) vs Pison
+// on a stream of JSON documents.
 //
 //   jsonbench --dataset twitter_small_records.json [options]
 //
@@ -25,6 +25,8 @@
 #include "pison_engine.h"
 #include "simdjson_engine.h"
 
+#include "simdjson.h"
+
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
@@ -44,8 +46,9 @@ struct options {
   std::string query_name;
   std::vector<size_t> threads;
   int reps = 3;
-  size_t slice_kb = 1024;   // parse_many_parallel slice size
+  size_t slice_kb = 64;     // parallel::parse_many slice size
   size_t batch_mb = 16;     // iterate_many batch for the built-in 2-thread mode
+  bool static_partition = true;   // --assign dynamic to override
   bool single_record = false;
   bool verify_only = false;
   size_t dump = 0;
@@ -54,10 +57,21 @@ struct options {
   // scripts narrow this so a slice-size or thread-count study does not re-run
   // the loader and agreement work every time.
   std::string sections = "load,verify,single,scaling,e2e";
+  // Restrict the scaling section to one engine. Aggregate profilers
+  // (perf stat -a) cannot attribute counters to an engine when several run in
+  // the same process, so isolating one is the only way to ask "what is *this*
+  // engine waiting on". The other sections are unaffected: pair this with
+  // --sections scaling so nothing else runs either.
+  std::string only_engine;
+  // Force a simdjson kernel (haswell, icelake, ...) instead of runtime dispatch.
+  std::string impl;
 
   bool wants(const char *s) const {
     return sections.find(s) != std::string::npos;
   }
+  bool wants_engine(const char *e) const {
+    return only_engine.empty() || only_engine == e;
+  }
 };
 
 void usage() {
@@ -69,14 +83,19 @@ void usage() {
       "                       (default: inferred from the filename)\n"
       "  --threads a,b,c      thread counts to sweep (default: 1..hw, doubling)\n"
       "  --reps <n>           repetitions per configuration, best wins (default 3)\n"
-      "  --slice-kb <n>       parse_many_parallel slice size (default 1024)\n"
+      "  --slice-kb <n>       parallel slice size (default 64)\n"
+      "  --assign <mode>      slice assignment: static (default) or dynamic\n"
       "  --batch-mb <n>       iterate_many batch size (default 16)\n"
       "  --single-record      treat the input as one bulky JSON document\n"
       "  --verify             check that the engines agree, then exit\n"
       "  --dump <n>           print the first n extracted values from each\n"
       "                       engine side by side, then exit\n"
       "  --sections <list>    comma list of load,verify,single,scaling,e2e,format\n"
-      "                       (default: all but format)\n");
+      "                       (default: all but format)\n"
+      "  --engine-only <name> restrict the scaling section to this engine (e.g.\n"
+      "                       simdjson-parallel); pair with --sections scaling\n"
+      "                       to profile one engine alone\n"
+      "  --impl <name>        force a simdjson kernel (haswell, icelake, ...)\n");
 }
 
 bool parse_args(int argc, char **argv, options &o) {
@@ -91,10 +110,21 @@ bool parse_args(int argc, char **argv, options &o) {
     else if (a == "--reps") { o.reps = atoi(next().c_str()); }
     else if (a == "--slice-kb") { o.slice_kb = strtoull(next().c_str(), nullptr, 10); }
     else if (a == "--batch-mb") { o.batch_mb = strtoull(next().c_str(), nullptr, 10); }
+    else if (a == "--assign") {
+      const std::string mode = next();
+      if (mode != "static" && mode != "dynamic") {
+        std::fprintf(stderr, "--assign takes static or dynamic, not '%s'\n",
+                     mode.c_str());
+        return false;
+      }
+      o.static_partition = (mode == "static");
+    }
     else if (a == "--single-record") { o.single_record = true; }
     else if (a == "--verify") { o.verify_only = true; }
     else if (a == "--dump") { o.dump = strtoull(next().c_str(), nullptr, 10); }
     else if (a == "--sections") { o.sections = next(); }
+    else if (a == "--engine-only") { o.only_engine = next(); }
+    else if (a == "--impl") { o.impl = next(); }
     else if (a == "--levels") { o.levels = atoi(next().c_str()); }
     else if (a == "--threads") {
       std::stringstream ss(next());
@@ -147,7 +177,8 @@ void emit(const char *engine, const char *phase, const char *workload_name,
                 s.branch_misses / (double(bytes) / 1024.0),
                 s.cache_misses / (double(bytes) / 1024.0));
   }
-  std::printf(" reps=%d\n", o.reps);
+  std::printf(" spread_pct=%.2f reps=%d", s.spread_pct(), o.reps);
+  std::printf("\n");
   std::fflush(stdout);
 }
 
@@ -157,6 +188,16 @@ int main(int argc, char **argv) {
   options o;
   if (!parse_args(argc, argv, o)) { usage(); return 1; }
 
+  if (!o.impl.empty()) {
+    auto wanted = simdjson::get_available_implementations()[o.impl];
+    if (wanted == nullptr || !wanted->supported_by_runtime_system()) {
+      std::fprintf(stderr, "simdjson implementation unavailable: %s\n",
+                   o.impl.c_str());
+      return 1;
+    }
+    simdjson::get_active_implementation() = wanted;
+  }
+
   query_id q;
   if (!o.query_name.empty()) {
     if (!query_from_name(o.query_name, q)) {
@@ -508,6 +549,7 @@ int main(int argc, char **argv) {
   // Thread scaling.
   // -----------------------------------------------------------------------
   for (size_t t : o.wants("scaling") ? o.threads : std::vector<size_t>{}) {
+    if (o.wants_engine("pison-parallel"))
     for (const auto &ph : pison_phases) {
       auto s = measure_parallel(o.reps, [&] {
         pison::run_stream(ptext, tbl, q, ph.w, levels, t);
@@ -516,20 +558,25 @@ int main(int argc, char **argv) {
       emit("pison-parallel", ph.phase, pison::workload_name(ph.w), t, o, label,
            bytes, docs, s, e);
     }
+    if (o.wants_engine("simdjson-parallel"))
     for (const auto &ph : sj_phases) {
       auto s = measure_parallel(o.reps, [&] {
-        sj::run_parallel(data, bytes, q, ph.w, t, o.slice_kb << 10);
+        sj::run_parallel(data, bytes, q, ph.w, t, o.slice_kb << 10,
+                         o.static_partition);
       });
-      extraction e = sj::run_parallel(data, bytes, q, ph.w, t, o.slice_kb << 10);
+      extraction e = sj::run_parallel(data, bytes, q, ph.w, t, o.slice_kb << 10,
+                                      o.static_partition);
       emit("simdjson-parallel", ph.phase, sj::workload_name(ph.w), t, o, label,
            bytes, docs, s, e);
     }
     // The same slicing rule, carrying a conventional DOM parser instead of the
     // on-demand one. Nothing in the decomposition knows what parses a document,
     // so this measures how much of our throughput comes from the slicing and
     // how much from on-demand parsing.
+    dom::static_partition_flag() = o.static_partition;
     for (auto lib : kDomLibraries) {
       if (!dom::available(lib)) { continue; }
+      if (!o.wants_engine(dom::engine_name(lib, true))) { continue; }
       auto ds = measure_parallel(o.reps, [&] {
         dom::run_parallel(lib, data, bytes, q, t, o.slice_kb << 10, dom_longest);
       });