diff --git a/CMakeLists.txt b/CMakeLists.txt index 1f5c403..0fd7e94 100644 --- a/CMakeLists.txt +++ b/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} diff --git a/README.md b/README.md index 5d05df9..3f0e840 100644 --- a/README.md +++ b/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 [options] (default: inferred from the filename) --threads a,b,c thread counts to sweep (default: 1..hw, doubling) --reps repetitions per configuration, best wins (default 3) - --slice-kb parse_many_parallel slice size (default 1024) + --slice-kb parallel slice size (default 64) + --assign slice assignment: static (default) or dynamic + --batch-mb iterate_many batch size (default 16) --sections 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 print the first n extracted values from each engine + --engine-only run only this engine in the scaling section + --impl force a simdjson kernel (haswell, icelake, ...) + --levels 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: diff --git a/src/dom_engine.h b/src/dom_engine.h index 42ff00d..cb5dff7 100644 --- a/src/dom_engine.h +++ b/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 diff --git a/src/dom_parallel.h b/src/dom_parallel.h index 03f54b9..f5a5ea8 100644 --- a/src/dom_parallel.h +++ b/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 @@ -41,6 +42,8 @@ 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 cursor{0}; + const bool stat = static_partition_flag(); + const size_t slices = (length + slice_bytes - 1) / slice_bytes; std::vector shards(threads); std::vector workers; workers.reserve(threads); @@ -48,9 +51,16 @@ extraction run_sliced(const char *data, size_t length, size_t threads, 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); diff --git a/src/main.cpp b/src/main.cpp index d79c9ae..c602fa2 100644 --- a/src/main.cpp +++ b/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 #include #include @@ -44,8 +46,9 @@ struct options { std::string query_name; std::vector 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 repetitions per configuration, best wins (default 3)\n" - " --slice-kb parse_many_parallel slice size (default 1024)\n" + " --slice-kb parallel slice size (default 64)\n" + " --assign slice assignment: static (default) or dynamic\n" " --batch-mb 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 print the first n extracted values from each\n" " engine side by side, then exit\n" " --sections comma list of load,verify,single,scaling,e2e,format\n" - " (default: all but format)\n"); + " (default: all but format)\n" + " --engine-only restrict the scaling section to this engine (e.g.\n" + " simdjson-parallel); pair with --sections scaling\n" + " to profile one engine alone\n" + " --impl 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{}) { + 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,11 +558,14 @@ 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); } @@ -528,8 +573,10 @@ int main(int argc, char **argv) { // 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); }); diff --git a/src/metrics.h b/src/metrics.h index 6a1034e..66c87ff 100644 --- a/src/metrics.h +++ b/src/metrics.h @@ -38,12 +38,21 @@ inline double cpu_seconds() { // One measured configuration. struct measurement { double seconds = 0; // wall clock, best of the repetitions + double worst_seconds = 0; // wall clock of the slowest repetition double cpu_seconds = 0; // CPU time of that same repetition, all threads double instructions = 0; // 0 when counters are unavailable or meaningless double cycles = 0; double branch_misses = 0; double cache_misses = 0; bool has_counters = false; + + // Spread between the slowest and fastest repetition, relative to the + // fastest, in percent. Reporting only the best hides how repeatable a + // configuration was, and repeatability varies enormously between machines. + double spread_pct() const { + if (seconds <= 0 || worst_seconds <= 0) { return 0; } + return 100.0 * (worst_seconds - seconds) / seconds; + } }; // Run `fn` `reps` times and keep the fastest repetition, with per-thread @@ -60,6 +69,7 @@ template measurement measure_single(int reps, Fn &&fn) { fn(); counters::event_count c = collector.end(); double cpu1 = cpu_seconds(); + if (c.elapsed_sec() > best.worst_seconds) { best.worst_seconds = c.elapsed_sec(); } if (c.elapsed_sec() < best.seconds) { best.seconds = c.elapsed_sec(); best.cpu_seconds = cpu1 - cpu0; @@ -87,6 +97,7 @@ template measurement measure_parallel(int reps, Fn &&fn) { auto t1 = std::chrono::steady_clock::now(); double cpu1 = cpu_seconds(); double s = std::chrono::duration(t1 - t0).count(); + if (s > best.worst_seconds) { best.worst_seconds = s; } if (s < best.seconds) { best.seconds = s; best.cpu_seconds = cpu1 - cpu0; diff --git a/src/parallel_stream.h b/src/parallel_stream.h new file mode 100644 index 0000000..6e75fb2 --- /dev/null +++ b/src/parallel_stream.h @@ -0,0 +1,189 @@ +// Parse a stream of JSON documents across many threads, on top of simdjson's +// on-demand interface. +// +// This lived in simdjson as an experimental header (PR #2788); it belongs in +// user code instead. Nothing here needs to be inside the library -- it is a +// slicing rule plus a thread pool over the public `iterate_many` API -- and +// keeping it out means callers can adapt the decomposition to their own +// pipeline rather than accept ours. +// +// Design. Each worker claims a slice index from one atomic counter and calls +// simdjson::slice_at, which snaps both ends to a delimiter so slices abut and +// never split a document. Each +// worker owns its parser and its output vector, so the hot path is lock-free. +// Results come back as one vector per worker ("shards"): values keep their +// order within a shard, but shards interleave with respect to the input. +// +// Formats. The slicing rule needs a delimiter that cannot occur inside a JSON +// value: a line feed for NDJSON, or a record separator (0x1E) for RFC 7464. +// Comma-delimited input is not supported, because a top-level comma can only be +// found by a serial structural scan. +// +// stream_format::newline_delimited, where available, additionally lets simdjson +// skip the tail of a partially read document without walking its structural +// characters. +#ifndef JSONBENCH_PARALLEL_STREAM_H +#define JSONBENCH_PARALLEL_STREAM_H + +#include "simdjson.h" + +#include +#include +#include +#include +#include + +namespace jsonbench { +namespace parallel { + +using simdjson::error_code; +using simdjson::stream_format; +using simdjson::SUCCESS; + +struct options { + size_t threads = 0; // 0 = hardware_concurrency() + size_t slice_bytes = 64u << 10; + stream_format format = stream_format::whitespace_delimited; + // Each worker owns one contiguous run of slices. The alternative, claiming + // the next free slice from a shared counter, scatters a worker's regions + // across the input and costs up to 1.55x; it is kept only for reproduction. + bool static_partition = true; +}; + +template class result { +public: + const std::vector> &shards() const noexcept { return _shards; } + size_t size() const noexcept { + size_t n = 0; + for (const auto &s : _shards) { n += s.size(); } + return n; + } + size_t errors() const noexcept { return _errors; } + error_code first_error() const noexcept { return _first_error; } + + std::vector> _shards{}; + size_t _errors{0}; + error_code _first_error{SUCCESS}; +}; + +namespace internal { + +inline bool splittable(stream_format f) { + return f == stream_format::whitespace_delimited || + f == stream_format::json_sequence +#if JSONBENCH_HAVE_NEWLINE_DELIMITED + || f == stream_format::newline_delimited +#endif + ; +} + +inline char delimiter_for(stream_format f) { + return f == stream_format::json_sequence ? char(0x1e) : '\n'; +} + +#if JSONBENCH_HAVE_NEWLINE_DELIMITED +using simdjson::slice_at; +#else +// simdjson::slice_at is not in released simdjson yet; same contract. +inline simdjson::padded_string_view slice_at(simdjson::padded_string_view data, + char delimiter, size_t block_size, + size_t index) noexcept { + if (block_size == 0 || index > data.size() / block_size) { return {}; } + const size_t raw_begin = index * block_size; + if (raw_begin >= data.size()) { return {}; } + auto snap = [&](size_t want) -> size_t { + if (want >= data.size()) { return data.size(); } + const void *p = std::memchr(data.data() + want, delimiter, data.size() - want); + return p ? size_t(static_cast(p) - data.data()) + 1 : data.size(); + }; + const size_t begin = (raw_begin == 0) ? 0 : snap(raw_begin); + const size_t end = snap(raw_begin + block_size); + if (begin >= end) { return {}; } + return simdjson::padded_string_view(data.data() + begin, end - begin, + data.capacity() - begin); +} +#endif + +} // namespace internal + +// Extract a value of type T from every document in `json`. +// +// `extract(doc, out)` returns SUCCESS to keep `out`, or an error to skip the +// document. Errors are counted, not fatal. +template +result parse_many(simdjson::padded_string_view json, F &&extract, + options opt = {}) { + result out; + const size_t length = json.size(); + const char delimiter = internal::delimiter_for(opt.format); + + size_t workers = opt.threads; + if (workers == 0) { + unsigned hw = std::thread::hardware_concurrency(); + workers = hw > 1 ? size_t(hw) : 1; + } + if (!internal::splittable(opt.format)) { workers = 1; } + const size_t slice_bytes = opt.slice_bytes ? opt.slice_bytes : (256u << 10); + + out._shards.resize(workers); + std::vector errors(workers, 0); + std::vector first(workers, SUCCESS); + std::atomic cursor{0}; + + std::vector pool; + pool.reserve(workers); + for (size_t w = 0; w < workers; w++) { + pool.emplace_back([&, w] { + simdjson::ondemand::parser parser; + parser.threaded = false; + std::vector &shard = out._shards[w]; + + const size_t slices = (length + slice_bytes - 1) / slice_bytes; + size_t next = slices * w / workers; + const size_t last = slices * (w + 1) / workers; + + for (;;) { + size_t i; + if (opt.static_partition) { + if (next >= last) { break; } + i = next++; + } else { + i = cursor.fetch_add(1, std::memory_order_relaxed); + } + if (i * slice_bytes >= length) { break; } + auto piece = internal::slice_at(json, delimiter, slice_bytes, i); + if (piece.empty()) { continue; } + + simdjson::ondemand::document_stream stream; + if (auto e = parser.iterate_many(piece, piece.size(), opt.format) + .get(stream)) { + errors[w]++; + if (!first[w]) { first[w] = e; } + continue; + } + for (auto it = stream.begin(); it != stream.end(); ++it) { + auto doc = *it; + T value; + if (auto e = extract(doc, value)) { + errors[w]++; + if (!first[w]) { first[w] = e; } + continue; + } + shard.push_back(std::move(value)); + } + } + }); + } + for (auto &t : pool) { t.join(); } + + for (size_t w = 0; w < workers; w++) { + out._errors += errors[w]; + if (!out._first_error) { out._first_error = first[w]; } + } + return out; +} + +} // namespace parallel +} // namespace jsonbench + +#endif diff --git a/src/simdjson_engine.cpp b/src/simdjson_engine.cpp index b08571e..7c12722 100644 --- a/src/simdjson_engine.cpp +++ b/src/simdjson_engine.cpp @@ -1,5 +1,6 @@ #include "simdjson_engine.h" +#include "parallel_stream.h" #include "simdjson.h" #include @@ -217,9 +218,16 @@ error_code dispatch(Doc doc, extraction &out, query_id q, workload w) { extraction run_serial(const char *data, size_t size, query_id q, workload w, bool threaded, size_t batch_bytes, std::vector *trace, size_t trace_limit) { + // The corpus is one document per line, the same structure Pison is handed as + // its record table, so we tell simdjson too. The format study calls + // run_serial_format directly and is unaffected by this. return run_serial_format(data, size, q, w, threaded, batch_bytes, - stream_format::whitespace_delimited, trace, - trace_limit); +#if JSONBENCH_HAVE_NEWLINE_DELIMITED + stream_format::newline_delimited, +#else + stream_format::whitespace_delimited, +#endif + trace, trace_limit); } extraction run_serial_format(const char *data, size_t size, query_id q, @@ -244,18 +252,28 @@ extraction run_serial_format(const char *data, size_t size, query_id q, } extraction run_parallel(const char *data, size_t size, query_id q, workload w, - size_t threads, size_t slice_bytes) { - experimental::parallel_stream_options options; + size_t threads, size_t slice_bytes, + bool static_partition) { + parallel::options options; options.threads = threads; options.slice_bytes = slice_bytes; + options.static_partition = static_partition; +#if JSONBENCH_HAVE_NEWLINE_DELIMITED + // Our corpus is strictly one document per line, which lets simdjson skip the + // tail of a partially read document instead of walking it. + options.format = stream_format::newline_delimited; +#else options.format = stream_format::whitespace_delimited; +#endif padded_string_view json(data, size, size + SIMDJSON_PADDING); - auto result = experimental::parse_many_parallel( + auto result = parallel::parse_many( json, [q, w](auto doc, extraction &out) { return dispatch(doc, out, q, w); }, options); extraction total; - result.for_each([&](const extraction &e) { total.merge(e); }); + for (const auto &shard : result.shards()) { + for (const extraction &e : shard) { total.merge(e); } + } return total; } diff --git a/src/simdjson_engine.h b/src/simdjson_engine.h index 3f74605..a380742 100644 --- a/src/simdjson_engine.h +++ b/src/simdjson_engine.h @@ -40,9 +40,10 @@ extraction run_serial_format(const char *data, size_t size, query_id q, std::vector *trace = nullptr, size_t trace_limit = 0); -// experimental::parse_many_parallel from simdjson PR #2788. +// This repository's own parallel stream driver, src/parallel_stream.h. extraction run_parallel(const char *data, size_t size, query_id q, workload w, - size_t threads, size_t slice_bytes); + size_t threads, size_t slice_bytes, + bool static_partition = true); const char *implementation_name();