admin/simdjson-simdjson
1
0
Fork 0
mirror of https://github.com/simdjson/simdjson synced 2026-06-08 17:27:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove document_stream from singlestage

Browse Source
This commit is contained in:
John Keiser
2023-07-17 16:31:37 -07:00
parent 91b74c593c
commit 7edd32dc6b
11 changed files with 149 additions and 1032 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/simdjson/generic/singlestage/amalgamated.h
-2
View File
@@ -17,7 +17,6 @@
#include "simdjson/generic/singlestage/array.h"
#include "simdjson/generic/singlestage/array_iterator.h"
#include "simdjson/generic/singlestage/document.h"
#include "simdjson/generic/singlestage/document_stream.h"
#include "simdjson/generic/singlestage/field.h"
#include "simdjson/generic/singlestage/object.h"
#include "simdjson/generic/singlestage/object_iterator.h"
@@ -27,7 +26,6 @@
#include "simdjson/generic/singlestage/array-inl.h"
#include "simdjson/generic/singlestage/array_iterator-inl.h"
#include "simdjson/generic/singlestage/document-inl.h"
#include "simdjson/generic/singlestage/document_stream-inl.h"
#include "simdjson/generic/singlestage/field-inl.h"
#include "simdjson/generic/singlestage/json_iterator-inl.h"
#include "simdjson/generic/singlestage/json_type-inl.h"
include/simdjson/generic/singlestage/base.h
-1
View File
@@ -27,7 +27,6 @@ class array;
class array_iterator;
class document;
class document_reference;
class document_stream;
class field;
class json_iterator;
enum class json_type;
include/simdjson/generic/singlestage/document.h
-1
View File
@@ -594,7 +594,6 @@ protected:
friend class array;
friend class field;
friend class token;
friend class document_stream;
friend class document_reference;
};
include/simdjson/generic/singlestage/document_stream-inl.h
-424
View File
@@ -1,424 +0,0 @@
#ifndef SIMDJSON_GENERIC_SINGLESTAGE_DOCUMENT_STREAM_INL_H
#ifndef SIMDJSON_AMALGAMATED
#define SIMDJSON_GENERIC_SINGLESTAGE_DOCUMENT_STREAM_INL_H
#include "simdjson/generic/singlestage/base.h"
#include "simdjson/generic/singlestage/document_stream.h"
#include "simdjson/generic/singlestage/document-inl.h"
#include "simdjson/generic/implementation_simdjson_result_base-inl.h"
#endif // SIMDJSON_AMALGAMATED
#include <algorithm>
#include <stdexcept>
namespace simdjson {
namespace SIMDJSON_IMPLEMENTATION {
namespace singlestage {
#ifdef SIMDJSON_THREADS_ENABLED
inline void stage1_worker::finish() {
// After calling "run" someone would call finish() to wait
// for the end of the processing.
// This function will wait until either the thread has done
// the processing or, else, the destructor has been called.
std::unique_lock<std::mutex> lock(locking_mutex);
cond_var.wait(lock, [this]{return has_work == false;});
}
inline stage1_worker::~stage1_worker() {
// The thread may never outlive the stage1_worker instance
// and will always be stopped/joined before the stage1_worker
// instance is gone.
stop_thread();
}
inline void stage1_worker::start_thread() {
std::unique_lock<std::mutex> lock(locking_mutex);
if(thread.joinable()) {
return; // This should never happen but we never want to create more than one thread.
}
thread = std::thread([this]{
while(true) {
std::unique_lock<std::mutex> thread_lock(locking_mutex);
// We wait for either "run" or "stop_thread" to be called.
cond_var.wait(thread_lock, [this]{return has_work || !can_work;});
// If, for some reason, the stop_thread() method was called (i.e., the
// destructor of stage1_worker is called, then we want to immediately destroy
// the thread (and not do any more processing).
if(!can_work) {
break;
}
this->owner->stage1_thread_error = this->owner->run_stage1(*this->stage1_thread_parser,
this->_next_batch_start);
this->has_work = false;
// The condition variable call should be moved after thread_lock.unlock() for performance
// reasons but thread sanitizers may report it as a data race if we do.
// See https://stackoverflow.com/questions/35775501/c-should-condition-variable-be-notified-under-lock
cond_var.notify_one(); // will notify "finish"
thread_lock.unlock();
}
}
);
}
inline void stage1_worker::stop_thread() {
std::unique_lock<std::mutex> lock(locking_mutex);
// We have to make sure that all locks can be released.
can_work = false;
has_work = false;
cond_var.notify_all();
lock.unlock();
if(thread.joinable()) {
thread.join();
}
}
inline void stage1_worker::run(document_stream * ds, parser * stage1, size_t next_batch_start) {
std::unique_lock<std::mutex> lock(locking_mutex);
owner = ds;
_next_batch_start = next_batch_start;
stage1_thread_parser = stage1;
has_work = true;
// The condition variable call should be moved after thread_lock.unlock() for performance
// reasons but thread sanitizers may report it as a data race if we do.
// See https://stackoverflow.com/questions/35775501/c-should-condition-variable-be-notified-under-lock
cond_var.notify_one(); // will notify the thread lock that we have work
lock.unlock();
}
#endif // SIMDJSON_THREADS_ENABLED
simdjson_inline document_stream::document_stream(
singlestage::parser &_parser,
const uint8_t *_buf,
size_t _len,
size_t _batch_size,
bool _allow_comma_separated
) noexcept
: parser{&_parser},
buf{_buf},
len{_len},
batch_size{_batch_size <= MINIMAL_BATCH_SIZE ? MINIMAL_BATCH_SIZE : _batch_size},
allow_comma_separated{_allow_comma_separated},
error{SUCCESS}
#ifdef SIMDJSON_THREADS_ENABLED
, use_thread(_parser.threaded) // we need to make a copy because _parser.threaded can change
#endif
{
#ifdef SIMDJSON_THREADS_ENABLED
if(worker.get() == nullptr) {
error = MEMALLOC;
}
#endif
}
simdjson_inline document_stream::document_stream() noexcept
: parser{nullptr},
buf{nullptr},
len{0},
batch_size{0},
allow_comma_separated{false},
error{UNINITIALIZED}
#ifdef SIMDJSON_THREADS_ENABLED
, use_thread(false)
#endif
{
}
simdjson_inline document_stream::~document_stream() noexcept
{
#ifdef SIMDJSON_THREADS_ENABLED
worker.reset();
#endif
}
inline size_t document_stream::size_in_bytes() const noexcept {
return len;
}
inline size_t document_stream::truncated_bytes() const noexcept {
if(error == CAPACITY) { return len - batch_start; }
return parser->implementation->structural_indexes[parser->implementation->n_structural_indexes] - parser->implementation->structural_indexes[parser->implementation->n_structural_indexes + 1];
}
simdjson_inline document_stream::iterator::iterator() noexcept
: stream{nullptr}, finished{true} {
}
simdjson_inline document_stream::iterator::iterator(document_stream* _stream, bool is_end) noexcept
: stream{_stream}, finished{is_end} {
}
simdjson_inline simdjson_result<singlestage::document_reference> document_stream::iterator::operator*() noexcept {
//if(stream->error) { return stream->error; }
return simdjson_result<singlestage::document_reference>(stream->doc, stream->error);
}
simdjson_inline document_stream::iterator& document_stream::iterator::operator++() noexcept {
// If there is an error, then we want the iterator
// to be finished, no matter what. (E.g., we do not
// keep generating documents with errors, or go beyond
// a document with errors.)
//
// Users do not have to call "operator*()" when they use operator++,
// so we need to end the stream in the operator++ function.
//
// Note that setting finished = true is essential otherwise
// we would enter an infinite loop.
if (stream->error) { finished = true; }
// Note that stream->error() is guarded against error conditions
// (it will immediately return if stream->error casts to false).
// In effect, this next function does nothing when (stream->error)
// is true (hence the risk of an infinite loop).
stream->next();
// If that was the last document, we're finished.
// It is the only type of error we do not want to appear
// in operator*.
if (stream->error == EMPTY) { finished = true; }
// If we had any other kind of error (not EMPTY) then we want
// to pass it along to the operator* and we cannot mark the result
// as "finished" just yet.
return *this;
}
simdjson_inline bool document_stream::iterator::operator!=(const document_stream::iterator &other) const noexcept {
return finished != other.finished;
}
simdjson_inline document_stream::iterator document_stream::begin() noexcept {
start();
// If there are no documents, we're finished.
return iterator(this, error == EMPTY);
}
simdjson_inline document_stream::iterator document_stream::end() noexcept {
return iterator(this, true);
}
inline void document_stream::start() noexcept {
if (error) { return; }
error = parser->allocate(batch_size);
if (error) { return; }
// Always run the first stage 1 parse immediately
batch_start = 0;
error = run_stage1(*parser, batch_start);
while(error == EMPTY) {
// In exceptional cases, we may start with an empty block
batch_start = next_batch_start();
if (batch_start >= len) { return; }
error = run_stage1(*parser, batch_start);
}
if (error) { return; }
doc_index = batch_start;
doc = document(json_iterator(&buf[batch_start], parser));
doc.iter._streaming = true;
#ifdef SIMDJSON_THREADS_ENABLED
if (use_thread && next_batch_start() < len) {
// Kick off the first thread on next batch if needed
error = stage1_thread_parser.allocate(batch_size);
if (error) { return; }
worker->start_thread();
start_stage1_thread();
if (error) { return; }
}
#endif // SIMDJSON_THREADS_ENABLED
}
inline void document_stream::next() noexcept {
// We always enter at once once in an error condition.
if (error) { return; }
next_document();
if (error) { return; }
auto cur_struct_index = doc.iter._root - parser->implementation->structural_indexes.get();
doc_index = batch_start + parser->implementation->structural_indexes[cur_struct_index];
// Check if at end of structural indexes (i.e. at end of batch)
if(cur_struct_index >= static_cast<int64_t>(parser->implementation->n_structural_indexes)) {
error = EMPTY;
// Load another batch (if available)
while (error == EMPTY) {
batch_start = next_batch_start();
if (batch_start >= len) { break; }
#ifdef SIMDJSON_THREADS_ENABLED
if(use_thread) {
load_from_stage1_thread();
} else {
error = run_stage1(*parser, batch_start);
}
#else
error = run_stage1(*parser, batch_start);
#endif
/**
* Whenever we move to another window, we need to update all pointers to make
* it appear as if the input buffer started at the beginning of the window.
*
* Take this input:
*
* {"z":5} {"1":1,"2":2,"4":4} [7, 10, 9] [15, 11, 12, 13] [154, 110, 112, 1311]
*
* Say you process the following window...
*
* '{"z":5} {"1":1,"2":2,"4":4} [7, 10, 9]'
*
* When you do so, the json_iterator has a pointer at the beginning of the memory region
* (pointing at the beginning of '{"z"...'.
*
* When you move to the window that starts at...
*
* '[7, 10, 9] [15, 11, 12, 13] ...
*
* then it is not sufficient to just run stage 1. You also need to re-anchor the
* json_iterator so that it believes we are starting at '[7, 10, 9]...'.
*
* Under the DOM front-end, this gets done automatically because the parser owns
* the pointer the data, and when you call stage1 and then stage2 on the same
* parser, then stage2 will run on the pointer acquired by stage1.
*
* That is, stage1 calls "this->buf = _buf" so the parser remembers the buffer that
* we used. But json_iterator has no callback when stage1 is called on the parser.
* In fact, I think that the parser is unaware of json_iterator.
*
*
* So we need to re-anchor the json_iterator after each call to stage 1 so that
* all of the pointers are in sync.
*/
doc.iter = json_iterator(&buf[batch_start], parser);
doc.iter._streaming = true;
/**
* End of resync.
*/
if (error) { continue; } // If the error was EMPTY, we may want to load another batch.
doc_index = batch_start;
}
}
}
inline void document_stream::next_document() noexcept {
// Go to next place where depth=0 (document depth)
error = doc.iter.skip_child(0);
if (error) { return; }
// Always set depth=1 at the start of document
doc.iter._depth = 1;
// consume comma if comma separated is allowed
if (allow_comma_separated) { doc.iter.consume_character(','); }
// Resets the string buffer at the beginning, thus invalidating the strings.
doc.iter._string_buf_loc = parser->string_buf.get();
doc.iter._root = doc.iter.position();
}
inline size_t document_stream::next_batch_start() const noexcept {
return batch_start + parser->implementation->structural_indexes[parser->implementation->n_structural_indexes];
}
inline error_code document_stream::run_stage1(singlestage::parser &p, size_t _batch_start) noexcept {
// This code only updates the structural index in the parser, it does not update any json_iterator
// instance.
size_t remaining = len - _batch_start;
if (remaining <= batch_size) {
return p.implementation->stage1(&buf[_batch_start], remaining, stage1_mode::streaming_final);
} else {
return p.implementation->stage1(&buf[_batch_start], batch_size, stage1_mode::streaming_partial);
}
}
simdjson_inline size_t document_stream::iterator::current_index() const noexcept {
return stream->doc_index;
}
simdjson_inline std::string_view document_stream::iterator::source() const noexcept {
auto depth = stream->doc.iter.depth();
auto cur_struct_index = stream->doc.iter._root - stream->parser->implementation->structural_indexes.get();
// If at root, process the first token to determine if scalar value
if (stream->doc.iter.at_root()) {
switch (stream->buf[stream->batch_start + stream->parser->implementation->structural_indexes[cur_struct_index]]) {
case '{': case '[': // Depth=1 already at start of document
break;
case '}': case ']':
depth--;
break;
default: // Scalar value document
// TODO: Remove any trailing whitespaces
// This returns a string spanning from start of value to the beginning of the next document (excluded)
return std::string_view(reinterpret_cast<const char*>(stream->buf) + current_index(), stream->parser->implementation->structural_indexes[++cur_struct_index] - current_index() - 1);
}
cur_struct_index++;
}
while (cur_struct_index <= static_cast<int64_t>(stream->parser->implementation->n_structural_indexes)) {
switch (stream->buf[stream->batch_start + stream->parser->implementation->structural_indexes[cur_struct_index]]) {
case '{': case '[':
depth++;
break;
case '}': case ']':
depth--;
break;
}
if (depth == 0) { break; }
cur_struct_index++;
}
return std::string_view(reinterpret_cast<const char*>(stream->buf) + current_index(), stream->parser->implementation->structural_indexes[cur_struct_index] - current_index() + stream->batch_start + 1);;
}
inline error_code document_stream::iterator::error() const noexcept {
return stream->error;
}
#ifdef SIMDJSON_THREADS_ENABLED
inline void document_stream::load_from_stage1_thread() noexcept {
worker->finish();
// Swap to the parser that was loaded up in the thread. Make sure the parser has
// enough memory to swap to, as well.
std::swap(stage1_thread_parser,*parser);
error = stage1_thread_error;
if (error) { return; }
// If there's anything left, start the stage 1 thread!
if (next_batch_start() < len) {
start_stage1_thread();
}
}
inline void document_stream::start_stage1_thread() noexcept {
// we call the thread on a lambda that will update
// this->stage1_thread_error
// there is only one thread that may write to this value
// TODO this is NOT exception-safe.
this->stage1_thread_error = UNINITIALIZED; // In case something goes wrong, make sure it's an error
size_t _next_batch_start = this->next_batch_start();
worker->run(this, & this->stage1_thread_parser, _next_batch_start);
}
#endif // SIMDJSON_THREADS_ENABLED
} // namespace singlestage
} // namespace SIMDJSON_IMPLEMENTATION
} // namespace simdjson
namespace simdjson {
simdjson_inline simdjson_result<SIMDJSON_IMPLEMENTATION::singlestage::document_stream>::simdjson_result(
error_code error
) noexcept :
implementation_simdjson_result_base<SIMDJSON_IMPLEMENTATION::singlestage::document_stream>(error)
{
}
simdjson_inline simdjson_result<SIMDJSON_IMPLEMENTATION::singlestage::document_stream>::simdjson_result(
SIMDJSON_IMPLEMENTATION::singlestage::document_stream &&value
) noexcept :
implementation_simdjson_result_base<SIMDJSON_IMPLEMENTATION::singlestage::document_stream>(
std::forward<SIMDJSON_IMPLEMENTATION::singlestage::document_stream>(value)
)
{
}
}
#endif // SIMDJSON_GENERIC_SINGLESTAGE_DOCUMENT_STREAM_INL_H
include/simdjson/generic/singlestage/document_stream.h
-339
View File
@@ -1,339 +0,0 @@
#ifndef SIMDJSON_GENERIC_SINGLESTAGE_DOCUMENT_STREAM_H
#ifndef SIMDJSON_AMALGAMATED
#define SIMDJSON_GENERIC_SINGLESTAGE_DOCUMENT_STREAM_H
#include "simdjson/generic/singlestage/base.h"
#include "simdjson/generic/implementation_simdjson_result_base.h"
#include "simdjson/generic/singlestage/document.h"
#include "simdjson/generic/singlestage/parser.h"
#endif // SIMDJSON_AMALGAMATED
#ifdef SIMDJSON_THREADS_ENABLED
#include <thread>
#include <mutex>
#include <condition_variable>
#endif
namespace simdjson {
namespace SIMDJSON_IMPLEMENTATION {
namespace singlestage {
#ifdef SIMDJSON_THREADS_ENABLED
/** @private Custom worker class **/
struct stage1_worker {
stage1_worker() noexcept = default;
stage1_worker(const stage1_worker&) = delete;
stage1_worker(stage1_worker&&) = delete;
stage1_worker operator=(const stage1_worker&) = delete;
~stage1_worker();
/**
* We only start the thread when it is needed, not at object construction, this may throw.
* You should only call this once.
**/
void start_thread();
/**
* Start a stage 1 job. You should first call 'run', then 'finish'.
* You must call start_thread once before.
*/
void run(document_stream * ds, parser * stage1, size_t next_batch_start);
/** Wait for the run to finish (blocking). You should first call 'run', then 'finish'. **/
void finish();
private:
/**
* Normally, we would never stop the thread. But we do in the destructor.
* This function is only safe assuming that you are not waiting for results. You
* should have called run, then finish, and be done.
**/
void stop_thread();
std::thread thread{};
/** These three variables define the work done by the thread. **/
singlestage::parser * stage1_thread_parser{};
size_t _next_batch_start{};
document_stream * owner{};
/**
* We have two state variables. This could be streamlined to one variable in the future but
* we use two for clarity.
*/
bool has_work{false};
bool can_work{true};
/**
* We lock using a mutex.
*/
std::mutex locking_mutex{};
std::condition_variable cond_var{};
friend class document_stream;
};
#endif // SIMDJSON_THREADS_ENABLED
/**
* A forward-only stream of documents.
*
* Produced by parser::iterate_many.
*
*/
class document_stream {
public:
/**
* Construct an uninitialized document_stream.
*
* ```c++
* document_stream docs;
* auto error = parser.iterate_many(json).get(docs);
* ```
*/
simdjson_inline document_stream() noexcept;
/** Move one document_stream to another. */
simdjson_inline document_stream(document_stream &&other) noexcept = default;
/** Move one document_stream to another. */
simdjson_inline document_stream &operator=(document_stream &&other) noexcept = default;
simdjson_inline ~document_stream() noexcept;
/**
* Returns the input size in bytes.
*/
inline size_t size_in_bytes() const noexcept;
/**
* After iterating through the stream, this method
* returns the number of bytes that were not parsed at the end
* of the stream. If truncated_bytes() differs from zero,
* then the input was truncated maybe because incomplete JSON
* documents were found at the end of the stream. You
* may need to process the bytes in the interval [size_in_bytes()-truncated_bytes(), size_in_bytes()).
*
* You should only call truncated_bytes() after streaming through all
* documents, like so:
*
* document_stream stream = parser.iterate_many(json,window);
* for(auto & doc : stream) {
* // do something with doc
* }
* size_t truncated = stream.truncated_bytes();
*
*/
inline size_t truncated_bytes() const noexcept;
class iterator {
public:
using value_type = simdjson_result<document>;
using reference = value_type;
using difference_type = std::ptrdiff_t;
using iterator_category = std::input_iterator_tag;
/**
* Default constructor.
*/
simdjson_inline iterator() noexcept;
/**
* Get the current document (or error).
*/
simdjson_inline simdjson_result<singlestage::document_reference> operator*() noexcept;
/**
* Advance to the next document (prefix).
*/
inline iterator& operator++() noexcept;
/**
* Check if we're at the end yet.
* @param other the end iterator to compare to.
*/
simdjson_inline bool operator!=(const iterator &other) const noexcept;
/**
* @private
*
* Gives the current index in the input document in bytes.
*
* document_stream stream = parser.parse_many(json,window);
* for(auto i = stream.begin(); i != stream.end(); ++i) {
* auto doc = *i;
* size_t index = i.current_index();
* }
*
* This function (current_index()) is experimental and the usage
* may change in future versions of simdjson: we find the API somewhat
* awkward and we would like to offer something friendlier.
*/
simdjson_inline size_t current_index() const noexcept;
/**
* @private
*
* Gives a view of the current document at the current position.
*
* document_stream stream = parser.iterate_many(json,window);
* for(auto i = stream.begin(); i != stream.end(); ++i) {
* std::string_view v = i.source();
* }
*
* The returned string_view instance is simply a map to the (unparsed)
* source string: it may thus include white-space characters and all manner
* of padding.
*
* This function (source()) is experimental and the usage
* may change in future versions of simdjson: we find the API somewhat
* awkward and we would like to offer something friendlier.
*
*/
simdjson_inline std::string_view source() const noexcept;
/**
* Returns error of the stream (if any).
*/
inline error_code error() const noexcept;
private:
simdjson_inline iterator(document_stream *s, bool finished) noexcept;
/** The document_stream we're iterating through. */
document_stream* stream;
/** Whether we're finished or not. */
bool finished;
friend class document;
friend class document_stream;
friend class json_iterator;
};
/**
* Start iterating the documents in the stream.
*/
simdjson_inline iterator begin() noexcept;
/**
* The end of the stream, for iterator comparison purposes.
*/
simdjson_inline iterator end() noexcept;
private:
document_stream &operator=(const document_stream &) = delete; // Disallow copying
document_stream(const document_stream &other) = delete; // Disallow copying
/**
* Construct a document_stream. Does not allocate or parse anything until the iterator is
* used.
*
* @param parser is a reference to the parser instance used to generate this document_stream
* @param buf is the raw byte buffer we need to process
* @param len is the length of the raw byte buffer in bytes
* @param batch_size is the size of the windows (must be strictly greater or equal to the largest JSON document)
*/
simdjson_inline document_stream(
singlestage::parser &parser,
const uint8_t *buf,
size_t len,
size_t batch_size,
bool allow_comma_separated
) noexcept;
/**
* Parse the first document in the buffer. Used by begin(), to handle allocation and
* initialization.
*/
inline void start() noexcept;
/**
* Parse the next document found in the buffer previously given to document_stream.
*
* The content should be a valid JSON document encoded as UTF-8. If there is a
* UTF-8 BOM, the caller is responsible for omitting it, UTF-8 BOM are
* discouraged.
*
* You do NOT need to pre-allocate a parser. This function takes care of
* pre-allocating a capacity defined by the batch_size defined when creating the
* document_stream object.
*
* The function returns simdjson::EMPTY if there is no more data to be parsed.
*
* The function returns simdjson::SUCCESS (as integer = 0) in case of success
* and indicates that the buffer has successfully been parsed to the end.
* Every document it contained has been parsed without error.
*
* The function returns an error code from simdjson/simdjson.h in case of failure
* such as simdjson::CAPACITY, simdjson::MEMALLOC, simdjson::DEPTH_ERROR and so forth;
* the simdjson::error_message function converts these error codes into a string).
*
* You can also check validity by calling parser.is_valid(). The same parser can
* and should be reused for the other documents in the buffer.
*/
inline void next() noexcept;
/** Move the json_iterator of the document to the location of the next document in the stream. */
inline void next_document() noexcept;
/** Get the next document index. */
inline size_t next_batch_start() const noexcept;
/** Pass the next batch through stage 1 with the given parser. */
inline error_code run_stage1(singlestage::parser &p, size_t batch_start) noexcept;
// Fields
singlestage::parser *parser;
const uint8_t *buf;
size_t len;
size_t batch_size;
bool allow_comma_separated;
/**
* We are going to use just one document instance. The document owns
* the json_iterator. It implies that we only ever pass a reference
* to the document to the users.
*/
document doc{};
/** The error (or lack thereof) from the current document. */
error_code error;
size_t batch_start{0};
size_t doc_index{};
#ifdef SIMDJSON_THREADS_ENABLED
/** Indicates whether we use threads. Note that this needs to be a constant during the execution of the parsing. */
bool use_thread;
inline void load_from_stage1_thread() noexcept;
/** Start a thread to run stage 1 on the next batch. */
inline void start_stage1_thread() noexcept;
/** Wait for the stage 1 thread to finish and capture the results. */
inline void finish_stage1_thread() noexcept;
/** The error returned from the stage 1 thread. */
error_code stage1_thread_error{UNINITIALIZED};
/** The thread used to run stage 1 against the next batch in the background. */
std::unique_ptr<stage1_worker> worker{new(std::nothrow) stage1_worker()};
/**
* The parser used to run stage 1 in the background. Will be swapped
* with the regular parser when finished.
*/
singlestage::parser stage1_thread_parser{};
friend struct stage1_worker;
#endif // SIMDJSON_THREADS_ENABLED
friend class parser;
friend class document;
friend class json_iterator;
friend struct simdjson_result<singlestage::document_stream>;
friend struct internal::simdjson_result_base<singlestage::document_stream>;
}; // document_stream
} // namespace singlestage
} // namespace SIMDJSON_IMPLEMENTATION
} // namespace simdjson
namespace simdjson {
template<>
struct simdjson_result<SIMDJSON_IMPLEMENTATION::singlestage::document_stream> : public SIMDJSON_IMPLEMENTATION::implementation_simdjson_result_base<SIMDJSON_IMPLEMENTATION::singlestage::document_stream> {
public:
simdjson_inline simdjson_result(SIMDJSON_IMPLEMENTATION::singlestage::document_stream &&value) noexcept; ///< @private
simdjson_inline simdjson_result(error_code error) noexcept; ///< @private
simdjson_inline simdjson_result() noexcept = default;
};
} // namespace simdjson
#endif // SIMDJSON_GENERIC_SINGLESTAGE_DOCUMENT_STREAM_H
include/simdjson/generic/singlestage/json_iterator.h
-1
View File
@@ -292,7 +292,6 @@ protected:
simdjson_inline token_position end() const noexcept;
friend class document;
friend class document_stream;
friend class object;
friend class array;
friend class value;
include/simdjson/generic/singlestage/parser-inl.h
+1 -16
View File
@@ -8,7 +8,7 @@
#include "simdjson/internal/dom_parser_implementation.h"
#include "simdjson/dom/base.h" // for MINIMAL_DOCUMENT_CAPACITY
#include "simdjson/generic/singlestage/base.h"
#include "simdjson/generic/singlestage/document_stream.h"
#include "simdjson/generic/singlestage/document.h"
#include "simdjson/generic/singlestage/parser.h"
#include "simdjson/generic/singlestage/raw_json_string.h"
#endif // SIMDJSON_AMALGAMATED
@@ -99,21 +99,6 @@ simdjson_warn_unused simdjson_inline simdjson_result<json_iterator> parser::iter
return json_iterator(reinterpret_cast<const uint8_t *>(json.data()), this);
}
inline simdjson_result<document_stream> parser::iterate_many(const uint8_t *buf, size_t len, size_t batch_size, bool allow_comma_separated) noexcept {
if(batch_size < MINIMAL_BATCH_SIZE) { batch_size = MINIMAL_BATCH_SIZE; }
if(allow_comma_separated && batch_size < len) { batch_size = len; }
return document_stream(*this, buf, len, batch_size, allow_comma_separated);
}
inline simdjson_result<document_stream> parser::iterate_many(const char *buf, size_t len, size_t batch_size, bool allow_comma_separated) noexcept {
return iterate_many(reinterpret_cast<const uint8_t *>(buf), len, batch_size, allow_comma_separated);
}
inline simdjson_result<document_stream> parser::iterate_many(const std::string &s, size_t batch_size, bool allow_comma_separated) noexcept {
return iterate_many(s.data(), s.length(), batch_size, allow_comma_separated);
}
inline simdjson_result<document_stream> parser::iterate_many(const padded_string &s, size_t batch_size, bool allow_comma_separated) noexcept {
return iterate_many(s.data(), s.length(), batch_size, allow_comma_separated);
}
simdjson_inline size_t parser::capacity() const noexcept {
return _capacity;
}
include/simdjson/generic/singlestage/parser.h
-100
View File
@@ -10,22 +10,6 @@ namespace simdjson {
namespace SIMDJSON_IMPLEMENTATION {
namespace singlestage {
/**
* The default batch size for document_stream instances for this On Demand kernel.
* Note that different On Demand kernel may use a different DEFAULT_BATCH_SIZE value
* in the future.
*/
static constexpr size_t DEFAULT_BATCH_SIZE = 1000000;
/**
* Some adversary might try to set the batch size to 0 or 1, which might cause problems.
* We set a minimum of 32B since anything else is highly likely to be an error. In practice,
* most users will want a much larger batch size.
*
* All non-negative MINIMAL_BATCH_SIZE values should be 'safe' except that, obviously, no JSON
* document can ever span 0 or 1 byte and that very large values would create memory allocation issues.
*/
static constexpr size_t MINIMAL_BATCH_SIZE = 32;
/**
* A JSON fragment iterator.
*
@@ -153,89 +137,6 @@ public:
*/
simdjson_warn_unused simdjson_result<json_iterator> iterate_raw(padded_string_view json) & noexcept;
/**
* Parse a buffer containing many JSON documents.
*
* auto json = R"({ "foo": 1 } { "foo": 2 } { "foo": 3 } )"_padded;
* singlestage::parser parser;
* singlestage::document_stream docs = parser.iterate_many(json);
* for (auto & doc : docs) {
* std::cout << doc["foo"] << std::endl;
* }
* // Prints 1 2 3
*
* No copy of the input buffer is made.
*
* The function is lazy: it may be that no more than one JSON document at a time is parsed.
*
* The caller is responsabile to ensure that the input string data remains unchanged and is
* not deleted during the loop.
*
* ### Format
*
* The buffer must contain a series of one or more JSON documents, concatenated into a single
* buffer, separated by ASCII whitespace. It effectively parses until it has a fully valid document,
* then starts parsing the next document at that point. (It does this with more parallelism and
* lookahead than you might think, though.)
*
* documents that consist of an object or array may omit the whitespace between them, concatenating
* with no separator. Documents that consist of a single primitive (i.e. documents that are not
* arrays or objects) MUST be separated with ASCII whitespace.
*
* The characters inside a JSON document, and between JSON documents, must be valid Unicode (UTF-8).
*
* The documents must not exceed batch_size bytes (by default 1MB) or they will fail to parse.
* Setting batch_size to excessively large or excessively small values may impact negatively the
* performance.
*
* ### REQUIRED: Buffer Padding
*
* The buffer must have at least SIMDJSON_PADDING extra allocated bytes. It does not matter what
* those bytes are initialized to, as long as they are allocated. These bytes will be read: if you
* using a sanitizer that verifies that no uninitialized byte is read, then you should initialize the
* SIMDJSON_PADDING bytes to avoid runtime warnings.
*
* ### Threads
*
* When compiled with SIMDJSON_THREADS_ENABLED, this method will use a single thread under the
* hood to do some lookahead.
*
* ### Parser Capacity
*
* If the parser's current capacity is less than batch_size, it will allocate enough capacity
* to handle it (up to max_capacity).
*
* @param buf The concatenated JSON to parse.
* @param len The length of the concatenated JSON.
* @param batch_size The batch size to use. MUST be larger than the largest document. The sweet
* spot is cache-related: small enough to fit in cache, yet big enough to
* parse as many documents as possible in one tight loop.
* Defaults to 10MB, which has been a reasonable sweet spot in our tests.
* @param allow_comma_separated (defaults on false) This allows a mode where the documents are
* separated by commas instead of whitespace. It comes with a performance
* penalty because the entire document is indexed at once (and the document must be
* less than 4 GB), and there is no multithreading. In this mode, the batch_size parameter
* is effectively ignored, as it is set to at least the document size.
* @return The stream, or an error. An empty input will yield 0 documents rather than an EMPTY error. Errors:
* - MEMALLOC if the parser does not have enough capacity and memory allocation fails
* - CAPACITY if the parser does not have enough capacity and batch_size > max_capacity.
* - other json errors if parsing fails. You should not rely on these errors to always the same for the
* same document: they may vary under runtime dispatch (so they may vary depending on your system and hardware).
*/
inline simdjson_result<document_stream> iterate_many(const uint8_t *buf, size_t len, size_t batch_size = DEFAULT_BATCH_SIZE, bool allow_comma_separated = false) noexcept;
/** @overload parse_many(const uint8_t *buf, size_t len, size_t batch_size) */
inline simdjson_result<document_stream> iterate_many(const char *buf, size_t len, size_t batch_size = DEFAULT_BATCH_SIZE, bool allow_comma_separated = false) noexcept;
/** @overload parse_many(const uint8_t *buf, size_t len, size_t batch_size) */
inline simdjson_result<document_stream> iterate_many(const std::string &s, size_t batch_size = DEFAULT_BATCH_SIZE, bool allow_comma_separated = false) noexcept;
inline simdjson_result<document_stream> iterate_many(const std::string &&s, size_t batch_size, bool allow_comma_separated = false) = delete;// unsafe
/** @overload parse_many(const uint8_t *buf, size_t len, size_t batch_size) */
inline simdjson_result<document_stream> iterate_many(const padded_string &s, size_t batch_size = DEFAULT_BATCH_SIZE, bool allow_comma_separated = false) noexcept;
inline simdjson_result<document_stream> iterate_many(const padded_string &&s, size_t batch_size, bool allow_comma_separated = false) = delete;// unsafe
/** @private We do not want to allow implicit conversion from C string to std::string. */
simdjson_result<document_stream> iterate_many(const char *buf, size_t batch_size = DEFAULT_BATCH_SIZE) noexcept = delete;
/** The capacity of this parser (the largest document it can process). */
simdjson_inline size_t capacity() const noexcept;
/** The maximum capacity of this parser (the largest document it is allowed to process). */
@@ -331,7 +232,6 @@ private:
#endif
friend class json_iterator;
friend class document_stream;
};
} // namespace singlestage
tests/singlestage/CMakeLists.txt
+2 -2
View File
@@ -9,7 +9,7 @@ add_cpp_test(singlestage_active_tests LABELS singlestage acceptance pe
add_cpp_test(singlestage_array_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_array_error_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_compilation_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_document_stream_tests LABELS singlestage acceptance per_implementation)
# add_cpp_test(singlestage_document_stream_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_error_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_error_location_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_json_pointer_tests LABELS singlestage acceptance per_implementation)
@@ -25,7 +25,7 @@ add_cpp_test(singlestage_readme_examples LABELS singlestage acceptance pe
add_cpp_test(singlestage_scalar_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_twitter_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_wrong_type_error_tests LABELS singlestage acceptance per_implementation)
add_cpp_test(singlestage_iterate_many_csv LABELS singlestage acceptance per_implementation)
# add_cpp_test(singlestage_iterate_many_csv LABELS singlestage acceptance per_implementation)
if(HAVE_POSIX_FORK AND HAVE_POSIX_WAIT) # assert tests use fork and wait, which aren't on MSVC
add_cpp_test(singlestage_assert_out_of_order_values LABELS assert per_implementation explicitonly singlestage)
tests/singlestage/singlestage_number_in_string_tests.cpp
+48 -48
View File
@@ -69,59 +69,59 @@ namespace number_in_string_tests {
TEST_SUCCEED();
}
bool stream_of_pure_int64() {
TEST_START();
auto json = R"( 1 2 3 )"_padded;
// bool stream_of_pure_int64() {
// TEST_START();
// auto json = R"( 1 2 3 )"_padded;
singlestage::parser parser;
singlestage::document_stream stream;
ASSERT_SUCCESS(parser.iterate_many(json).get(stream));
int document_index = 0;
for (auto doc : stream) {
document_index++;
int64_t result;
ASSERT_SUCCESS(doc.get_int64().get(result));
ASSERT_EQUAL(result,document_index);
}
ASSERT_EQUAL(3,document_index);
TEST_SUCCEED();
}
// singlestage::parser parser;
// singlestage::document_stream stream;
// ASSERT_SUCCESS(parser.iterate_many(json).get(stream));
// int document_index = 0;
// for (auto doc : stream) {
// document_index++;
// int64_t result;
// ASSERT_SUCCESS(doc.get_int64().get(result));
// ASSERT_EQUAL(result,document_index);
// }
// ASSERT_EQUAL(3,document_index);
// TEST_SUCCEED();
// }
bool stream_of_direct_int64() {
TEST_START();
auto json = R"( "1" "2" "3" )"_padded;
// bool stream_of_direct_int64() {
// TEST_START();
// auto json = R"( "1" "2" "3" )"_padded;
singlestage::parser parser;
singlestage::document_stream stream;
ASSERT_SUCCESS(parser.iterate_many(json).get(stream));
int document_index = 0;
for (auto doc : stream) {
document_index++;
int64_t result;
ASSERT_SUCCESS(doc.get_int64_in_string().get(result));
ASSERT_EQUAL(result,document_index);
}
ASSERT_EQUAL(3,document_index);
TEST_SUCCEED();
}
// singlestage::parser parser;
// singlestage::document_stream stream;
// ASSERT_SUCCESS(parser.iterate_many(json).get(stream));
// int document_index = 0;
// for (auto doc : stream) {
// document_index++;
// int64_t result;
// ASSERT_SUCCESS(doc.get_int64_in_string().get(result));
// ASSERT_EQUAL(result,document_index);
// }
// ASSERT_EQUAL(3,document_index);
// TEST_SUCCEED();
// }
bool stream_of_int64() {
TEST_START();
auto json = R"( ["1"] ["2"] ["3"] )"_padded;
// bool stream_of_int64() {
// TEST_START();
// auto json = R"( ["1"] ["2"] ["3"] )"_padded;
singlestage::parser parser;
singlestage::document_stream stream;
ASSERT_SUCCESS(parser.iterate_many(json).get(stream));
int document_index = 0;
for (auto doc : stream) {
document_index++;
int64_t result;
ASSERT_SUCCESS(doc.get_array().at(0).get_int64_in_string().get(result));
ASSERT_EQUAL(result,document_index);
}
ASSERT_EQUAL(3,document_index);
TEST_SUCCEED();
}
// singlestage::parser parser;
// singlestage::document_stream stream;
// ASSERT_SUCCESS(parser.iterate_many(json).get(stream));
// int document_index = 0;
// for (auto doc : stream) {
// document_index++;
// int64_t result;
// ASSERT_SUCCESS(doc.get_array().at(0).get_int64_in_string().get(result));
// ASSERT_EQUAL(result,document_index);
// }
// ASSERT_EQUAL(3,document_index);
// TEST_SUCCEED();
// }
bool array_double() {
TEST_START();
tests/singlestage/singlestage_readme_examples.cpp
+98 -98
View File
@@ -799,92 +799,92 @@ bool json_pointer_rewind() {
TEST_SUCCEED();
}
bool iterate_many_example() {
TEST_START();
auto json = R"([1,2,3] {"1":1,"2":3,"4":4} [1,2,3] )"_padded;
simdjson::singlestage::parser parser;
simdjson::singlestage::document_stream stream;
ASSERT_SUCCESS(parser.iterate_many(json).get(stream));
auto i = stream.begin();
size_t count{0};
size_t expected_indexes[3] = {0,9,29};
std::string_view expected_doc[3] = {"[1,2,3]", R"({"1":1,"2":3,"4":4})", "[1,2,3]"};
for(; i != stream.end(); ++i) {
auto doc = *i;
ASSERT_SUCCESS(doc.type());
ASSERT_SUCCESS(i.error());
ASSERT_EQUAL(i.current_index(),expected_indexes[count]);
ASSERT_EQUAL(i.source(),expected_doc[count]);
count++;
}
TEST_SUCCEED();
}
// bool iterate_many_example() {
// TEST_START();
// auto json = R"([1,2,3] {"1":1,"2":3,"4":4} [1,2,3] )"_padded;
// simdjson::singlestage::parser parser;
// simdjson::singlestage::document_stream stream;
// ASSERT_SUCCESS(parser.iterate_many(json).get(stream));
// auto i = stream.begin();
// size_t count{0};
// size_t expected_indexes[3] = {0,9,29};
// std::string_view expected_doc[3] = {"[1,2,3]", R"({"1":1,"2":3,"4":4})", "[1,2,3]"};
// for(; i != stream.end(); ++i) {
// auto doc = *i;
// ASSERT_SUCCESS(doc.type());
// ASSERT_SUCCESS(i.error());
// ASSERT_EQUAL(i.current_index(),expected_indexes[count]);
// ASSERT_EQUAL(i.source(),expected_doc[count]);
// count++;
// }
// TEST_SUCCEED();
// }
std::string my_string(singlestage::document& doc) {
std::stringstream ss;
ss << doc;
return ss.str();
}
// std::string my_string(singlestage::document& doc) {
// std::stringstream ss;
// ss << doc;
// return ss.str();
// }
bool iterate_many_truncated_example() {
TEST_START();
auto json = R"([1,2,3] {"1":1,"2":3,"4":4} {"key":"intentionally unclosed string )"_padded;
simdjson::singlestage::parser parser;
simdjson::singlestage::document_stream stream;
ASSERT_SUCCESS( parser.iterate_many(json,json.size()).get(stream) );
std::string_view expected[2] = {"[1,2,3]", R"({"1":1,"2":3,"4":4})"};
size_t count{0};
for(auto i = stream.begin(); i != stream.end(); ++i) {
ASSERT_EQUAL(i.source(),expected[count++]);
}
size_t truncated = stream.truncated_bytes();
ASSERT_EQUAL(truncated,39);
TEST_SUCCEED();
}
// bool iterate_many_truncated_example() {
// TEST_START();
// auto json = R"([1,2,3] {"1":1,"2":3,"4":4} {"key":"intentionally unclosed string )"_padded;
// simdjson::singlestage::parser parser;
// simdjson::singlestage::document_stream stream;
// ASSERT_SUCCESS( parser.iterate_many(json,json.size()).get(stream) );
// std::string_view expected[2] = {"[1,2,3]", R"({"1":1,"2":3,"4":4})"};
// size_t count{0};
// for(auto i = stream.begin(); i != stream.end(); ++i) {
// ASSERT_EQUAL(i.source(),expected[count++]);
// }
// size_t truncated = stream.truncated_bytes();
// ASSERT_EQUAL(truncated,39);
// TEST_SUCCEED();
// }
bool ndjson_basics_example() {
TEST_START();
auto json = R"({ "foo": 1 } { "foo": 2 } { "foo": 3 } )"_padded;
singlestage::parser parser;
singlestage::document_stream docs;
ASSERT_SUCCESS( parser.iterate_many(json).get(docs) );
size_t count{0};
int64_t expected[3] = {1,2,3};
for (auto doc : docs) {
int64_t actual{};
ASSERT_SUCCESS( doc["foo"].get(actual) );
ASSERT_EQUAL( actual, expected[count++] );
}
TEST_SUCCEED();
}
bool stream_capacity_example() {
auto json = R"([1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99,100])"_padded;
singlestage::parser parser;
singlestage::document_stream stream;
size_t counter{0};
auto error = parser.iterate_many(json, 50).get(stream);
if( error ) { /* handle the error */ }
for (auto doc: stream) {
if(counter < 6) {
int64_t val{};
error = doc.at_pointer("/4").get(val);
if( error ) { /* handle the error */ }
std::cout << "5 = " << val << std::endl;
} else {
singlestage::value val;
error = doc.at_pointer("/4").get(val);
// error == simdjson::CAPACITY
if(error) {
std::cerr << error << std::endl;
// We left 293 bytes unprocessed at the tail end of the input.
std::cout << " unprocessed bytes at the end: " << stream.truncated_bytes() << std::endl;
break;
}
}
counter++;
}
return true;
}
// bool ndjson_basics_example() {
// TEST_START();
// auto json = R"({ "foo": 1 } { "foo": 2 } { "foo": 3 } )"_padded;
// singlestage::parser parser;
// singlestage::document_stream docs;
// ASSERT_SUCCESS( parser.iterate_many(json).get(docs) );
// size_t count{0};
// int64_t expected[3] = {1,2,3};
// for (auto doc : docs) {
// int64_t actual{};
// ASSERT_SUCCESS( doc["foo"].get(actual) );
// ASSERT_EQUAL( actual, expected[count++] );
// }
// TEST_SUCCEED();
// }
// bool stream_capacity_example() {
// auto json = R"([1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5] [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99,100])"_padded;
// singlestage::parser parser;
// singlestage::document_stream stream;
// size_t counter{0};
// auto error = parser.iterate_many(json, 50).get(stream);
// if( error ) { /* handle the error */ }
// for (auto doc: stream) {
// if(counter < 6) {
// int64_t val{};
// error = doc.at_pointer("/4").get(val);
// if( error ) { /* handle the error */ }
// std::cout << "5 = " << val << std::endl;
// } else {
// singlestage::value val;
// error = doc.at_pointer("/4").get(val);
// // error == simdjson::CAPACITY
// if(error) {
// std::cerr << error << std::endl;
// // We left 293 bytes unprocessed at the tail end of the input.
// std::cout << " unprocessed bytes at the end: " << stream.truncated_bytes() << std::endl;
// break;
// }
// }
// counter++;
// }
// return true;
// }
@@ -1038,21 +1038,21 @@ int load_example_except_morecomplete(void) {
return EXIT_SUCCESS;
}
bool allow_comma_separated_example() {
TEST_START();
auto json = R"( 1, 2, 3, 4, "a", "b", "c", {"hello": "world"} , [1, 2, 3])"_padded;
singlestage::parser parser;
singlestage::document_stream doc_stream;
// We pass '32' as the batch size, but it is a bogus parameter because, since
// we pass 'true' to the allow_comma parameter, the batch size will be set to at least
// the document size.
auto error = parser.iterate_many(json, 32, true).get(doc_stream);
if(error) { std::cerr << error << std::endl; return false; }
for (auto doc : doc_stream) {
std::cout << doc.type() << std::endl;
}
TEST_SUCCEED();
}
// bool allow_comma_separated_example() {
// TEST_START();
// auto json = R"( 1, 2, 3, 4, "a", "b", "c", {"hello": "world"} , [1, 2, 3])"_padded;
// singlestage::parser parser;
// singlestage::document_stream doc_stream;
// // We pass '32' as the batch size, but it is a bogus parameter because, since
// // we pass 'true' to the allow_comma parameter, the batch size will be set to at least
// // the document size.
// auto error = parser.iterate_many(json, 32, true).get(doc_stream);
// if(error) { std::cerr << error << std::endl; return false; }
// for (auto doc : doc_stream) {
// std::cout << doc.type() << std::endl;
// }
// TEST_SUCCEED();
// }
#endif
bool test_load_example() {
TEST_START();