Introducing concurrency mode in JsonStream. (#373)

* JsonStream threaded prototype

* JsonStream Threaded version working. Still supporting non-threaded version.

* Fix where invalid files would enter infinite loop.

* SingleHeader update

* I will remove -pthread in cmake for now.

* Attempt at resolving the -pthread issue

.gitignore

@@ -1,4 +1,5 @@
 .vscode/
+.idea/
 /VS/
 /build/
 /benchbranch/

CMakeLists.txt

@@ -11,6 +11,9 @@ option(SIMDJSON_DISABLE_AVX "Forcefully disable AVX even if hardware supports it
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_MACOSX_RPATH OFF)
+set(CMAKE_THREAD_PREFER_PTHREAD ON)
+set(THREADS_PREFER_PTHREAD_FLAG ON)
+
 if (NOT CMAKE_BUILD_TYPE)
                 message(STATUS "No build type selected, default to Release")
                 set(CMAKE_BUILD_TYPE Release CACHE STRING "Choose the type of build." FORCE)
@@ -35,6 +38,11 @@ set(CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/tools/cmake")
 
 find_package(CTargets)
 find_package(Options)
+find_package(Threads REQUIRED)
+
+if(CMAKE_USE_PTHREADS_INIT)
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread")
+endif()
 
 install(DIRECTORY include/${SIMDJSON_LIB_NAME} DESTINATION include)
 set (TEST_DATA_DIR "${CMAKE_CURRENT_SOURCE_DIR}/jsonchecker/")

Makefile

@@ -22,7 +22,7 @@ else
 ARCHFLAGS ?= -msse4.2 -mpclmul # lowest supported feature set?
 endif
 
-CXXFLAGS = $(ARCHFLAGS) -std=c++17   -Wall -Wextra -Wshadow -Iinclude -Isrc -Ibenchmark/linux $(EXTRAFLAGS)
+CXXFLAGS = $(ARCHFLAGS) -std=c++17  -pthread -Wall -Wextra -Wshadow -Iinclude -Isrc -Ibenchmark/linux $(EXTRAFLAGS)
 CFLAGS =  $(ARCHFLAGS)  -Idependencies/ujson4c/3rdparty -Idependencies/ujson4c/src $(EXTRAFLAGS)
 
 

amalgamation.sh

@@ -181,16 +181,16 @@ echo "Giving final instructions:"
 CPPBIN=${DEMOCPP%%.*}
 
 echo "Try :"
-echo "c++ -O3 -std=c++17 -o ${CPPBIN} ${DEMOCPP}  && ./${CPPBIN} ../jsonexamples/twitter.json ../jsonexamples/amazon_cellphones.ndjson"
+echo "c++ -O3 -std=c++17 -pthread -o ${CPPBIN} ${DEMOCPP}  && ./${CPPBIN} ../jsonexamples/twitter.json ../jsonexamples/amazon_cellphones.ndjson"
 
 SINGLEHDR=$SCRIPTPATH/singleheader
 echo "Copying files to $SCRIPTPATH/singleheader "
 mkdir -p $SINGLEHDR
-echo "c++ -O3 -std=c++17 -o ${CPPBIN} ${DEMOCPP}  && ./${CPPBIN} ../jsonexamples/twitter.json ../jsonexamples/amazon_cellphones.ndjson" > $SINGLEHDR/README.md
+echo "c++ -O3 -std=c++17 -pthread -o ${CPPBIN} ${DEMOCPP}  && ./${CPPBIN} ../jsonexamples/twitter.json ../jsonexamples/amazon_cellphones.ndjson" > $SINGLEHDR/README.md
 cp ${AMAL_C} ${AMAL_H}  ${DEMOCPP} $SINGLEHDR
 ls $SINGLEHDR
 
-cd $SINGLEHDR && c++ -O3 -std=c++17 -o ${CPPBIN} ${DEMOCPP}  && ./${CPPBIN} ../jsonexamples/twitter.json ../jsonexamples/amazon_cellphones.ndjson
+cd $SINGLEHDR && c++ -O3 -std=c++17 -pthread -o ${CPPBIN} ${DEMOCPP}  && ./${CPPBIN} ../jsonexamples/twitter.json ../jsonexamples/amazon_cellphones.ndjson
 
 lowercase(){
     echo "$1" | tr 'A-Z' 'a-z'

benchmark/CMakeLists.txt

@@ -7,4 +7,7 @@ target_include_directories(${SIMDJSON_LIB_NAME}
 add_cpp_benchmark(parse)
 add_cpp_benchmark(statisticalmodel)
 add_cpp_benchmark(parse_stream)
+
+target_link_libraries(parse_stream Threads::Threads)
+
 add_executable(perfdiff perfdiff.cpp)

benchmark/parse_stream.cpp

@@ -8,7 +8,7 @@
 #include "simdjson/parsedjson.h"
 
 #define NB_ITERATION 5
-#define MIN_BATCH_SIZE 100000
+#define MIN_BATCH_SIZE 200000
 #define MAX_BATCH_SIZE 10000000
 
 bool test_baseline = false;
@@ -74,7 +74,7 @@ if(test_per_batch) {
     std::wclog << "Jsonstream: Speed per batch_size... from " << MIN_BATCH_SIZE
                << " bytes to " << MAX_BATCH_SIZE << " bytes..." << std::endl;
     std::cout << "Batch Size\t" << "Gigabytes/second\t" << "Nb of documents parsed" << std::endl;
-    for (size_t i = MIN_BATCH_SIZE; i <= MAX_BATCH_SIZE; i += (MAX_BATCH_SIZE - MIN_BATCH_SIZE) / 200) {
+    for (size_t i = MIN_BATCH_SIZE; i <= MAX_BATCH_SIZE; i += (MAX_BATCH_SIZE - MIN_BATCH_SIZE) / 30) {
         batch_size_res.insert(std::pair<size_t, double>(i, 0));
         int count;
         for (size_t j = 0; j < 5; j++) {

include/simdjson/jsonstream.h

@@ -3,6 +3,7 @@
 
 
 #include <algorithm>
+#include <thread>
 #include "simdjson/stage1_find_marks.h"
 #include "simdjson/stage2_build_tape.h"
 #include "simdjson/simdjson.h"
@@ -35,8 +36,17 @@ namespace simdjson {
         bool error_on_last_attempt{false};
         bool load_next_batch{true};
         size_t current_buffer_loc{0};
+        size_t last_json_buffer_loc{0};
+        size_t thread_current_buffer_loc{0};
         size_t n_parsed_docs{0};
         size_t n_bytes_parsed{0};
+
+        std::thread stage_1_thread;
+        simdjson::ParsedJson pj_thread;
+
+#ifdef SIMDJSON_THREADS_ENABLED
+        size_t find_last_json(const ParsedJson &pj);
+#endif
     };
 
 }

include/simdjson/portability.h

@@ -42,6 +42,11 @@
 #define TARGET_WESTMERE TARGET_REGION("sse4.2,pclmul")
 #define TARGET_ARM64
 
+// Is threading enabled?
+#if defined(BOOST_HAS_THREADS) || defined(_REENTRANT) || defined(_MT)
+#define SIMDJSON_THREADS_ENABLED 1
+#endif
+
 #ifdef _MSC_VER
 #include <intrin.h>
 #else

singleheader/README.md

@@ -1 +1 @@
-c++ -O3 -std=c++17 -o amalgamation_demo amalgamation_demo.cpp  && ./amalgamation_demo ../jsonexamples/twitter.json ../jsonexamples/amazon_cellphones.ndjson
+c++ -O3 -std=c++17 -pthread -o amalgamation_demo amalgamation_demo.cpp  && ./amalgamation_demo ../jsonexamples/twitter.json ../jsonexamples/amazon_cellphones.ndjson

singleheader/amalgamation_demo.cpp

@@ -1,4 +1,4 @@
-/* auto-generated on Thu 07 Nov 2019 05:05:37 PM EST. Do not edit! */
+/* auto-generated on Wed Nov 20 11:15:43 EST 2019. Do not edit! */
 
 #include <iostream>
 #include "simdjson.h"

singleheader/simdjson.h

@@ -1,4 +1,4 @@
-/* auto-generated on Thu 07 Nov 2019 05:05:37 PM EST. Do not edit! */
+/* auto-generated on Wed Nov 20 11:15:43 EST 2019. Do not edit! */
 /* begin file include/simdjson/simdjson_version.h */
 // /include/simdjson/simdjson_version.h automatically generated by release.py,
 // do not change by hand
@@ -59,6 +59,11 @@ enum {
 #define TARGET_WESTMERE TARGET_REGION("sse4.2,pclmul")
 #define TARGET_ARM64
 
+// Is threading enabled?
+#if defined(BOOST_HAS_THREADS) || defined(_REENTRANT) || defined(_MT)
+#define SIMDJSON_THREADS_ENABLED 1
+#endif
+
 #ifdef _MSC_VER
 #include <intrin.h>
 #else
@@ -705,6 +710,7 @@ const std::string &error_message(const int);
 #define SIMDJSON_PADDING_STRING_H
 #include <cstring>
 #include <memory>
+#include <string>
 
 namespace simdjson {
 // low-level function to allocate memory with padding so we can read passed the
@@ -743,6 +749,15 @@ public:
       : viable_size(o.viable_size), data_ptr(o.data_ptr) {
     o.data_ptr = nullptr; // we take ownership
   }
+
+  padded_string &operator=(padded_string &&o) {
+    data_ptr = o.data_ptr;
+    viable_size = o.viable_size;
+    o.data_ptr = nullptr; // we take ownership
+    o.viable_size = 0;
+    return *this;
+  }
+
   void swap(padded_string &o) {
     size_t tmp_viable_size = viable_size;
     char *tmp_data_ptr = data_ptr;
@@ -859,6 +874,7 @@ public:
   ParsedJson();
   ~ParsedJson();
   ParsedJson(ParsedJson &&p);
+  ParsedJson &operator=(ParsedJson &&o);
 
   // if needed, allocate memory so that the object is able to process JSON
   // documents having up to len bytes and max_depth "depth"
@@ -2013,6 +2029,7 @@ inline ParsedJson build_parsed_json(const padded_string &s) {
 
 
 #include <algorithm>
+#include <thread>
 
 namespace simdjson {
 
@@ -2042,8 +2059,17 @@ namespace simdjson {
         bool error_on_last_attempt{false};
         bool load_next_batch{true};
         size_t current_buffer_loc{0};
+        size_t last_json_buffer_loc{0};
+        size_t thread_current_buffer_loc{0};
         size_t n_parsed_docs{0};
         size_t n_bytes_parsed{0};
+
+        std::thread stage_1_thread;
+        simdjson::ParsedJson pj_thread;
+
+#ifdef SIMDJSON_THREADS_ENABLED
+        size_t find_last_json(const ParsedJson &pj);
+#endif
     };
 
 }

src/jsonstream.cpp

@@ -3,15 +3,14 @@
 #include <map>
 
 using namespace simdjson;
-
 void find_the_best_supported_implementation();
+
 typedef int (*stage1_functype)(const char *buf, size_t len, ParsedJson &pj, bool streaming);
 typedef int (*stage2_functype)(const char *buf, size_t len, ParsedJson &pj, size_t &next_json);
 
 stage1_functype best_stage1;
 stage2_functype best_stage2;
 
-
 JsonStream::JsonStream(const char *buf, size_t len, size_t batchSize)
         : _buf(buf), _len(len), _batch_size(batchSize) {
     find_the_best_supported_implementation();
@@ -30,11 +29,10 @@ void JsonStream::set_new_buffer(const char *buf, size_t len) {
 }
 
 int JsonStream::json_parse(ParsedJson &pj) {
-    //return json_parse_ptr.load(std::memory_order_relaxed)(buf, len, batch_size, pj, realloc_if_needed);
-
     if (pj.byte_capacity == 0) {
         const bool allocok = pj.allocate_capacity(_batch_size, _batch_size);
-        if (!allocok) {
+        const bool allocok_thread = pj_thread.allocate_capacity(_batch_size, _batch_size);
+        if (!allocok || !allocok_thread) {
             std::cerr << "can't allocate memory" << std::endl;
             return false;
         }
@@ -42,31 +40,64 @@ int JsonStream::json_parse(ParsedJson &pj) {
     else if (pj.byte_capacity < _batch_size) {
         return simdjson::CAPACITY;
     }
-
-    //Quick heuristic to see if it's worth parsing the remaining data in the batch
-    if(!load_next_batch && n_bytes_parsed > 0) {
-        const auto remaining_data = _batch_size - current_buffer_loc;
-        const auto avg_doc_len = (float) n_bytes_parsed / n_parsed_docs;
-
-        if(remaining_data < avg_doc_len)
-            load_next_batch = true;
-    }
+#ifdef SIMDJSON_THREADS_ENABLED
+    if(current_buffer_loc == last_json_buffer_loc)
+        load_next_batch = true;
+#endif
 
     if (load_next_batch){
+#ifdef SIMDJSON_THREADS_ENABLED
+        //First time loading
+        if(!stage_1_thread.joinable()){
+            _buf = &_buf[current_buffer_loc];
+            _len -= current_buffer_loc;
+            n_bytes_parsed += current_buffer_loc;
 
+            _batch_size = std::min(_batch_size, _len);
+            int stage1_is_ok = (*best_stage1)(_buf, _batch_size, pj, true);
+
+            if (stage1_is_ok != simdjson::SUCCESS) {
+                pj.error_code = stage1_is_ok;
+                return pj.error_code;
+            }
+        }
+
+            //the second thread is running or done.
+        else{
+            stage_1_thread.join();
+            std::swap(pj.structural_indexes, pj_thread.structural_indexes);
+            pj.n_structural_indexes = pj_thread.n_structural_indexes;
+
+            _buf = &_buf[last_json_buffer_loc];
+            _len -= last_json_buffer_loc;
+            n_bytes_parsed += last_json_buffer_loc;
+            last_json_buffer_loc = 0; //because we want to use it in the if above.
+        }
+
+        if(_len-_batch_size > 0) {
+            last_json_buffer_loc = find_last_json(pj);
+            _batch_size = std::min(_batch_size, _len - last_json_buffer_loc);
+            if(_batch_size>0)
+                stage_1_thread = std::thread(
+                        static_cast<stage1_functype>(*best_stage1),
+                        &_buf[last_json_buffer_loc], _batch_size,
+                        std::ref(pj_thread),
+                        true);
+
+        }
+#else
         _buf = &_buf[current_buffer_loc];
         _len -= current_buffer_loc;
         n_bytes_parsed += current_buffer_loc;
 
         _batch_size = std::min(_batch_size, _len);
-
         int stage1_is_ok = (*best_stage1)(_buf, _batch_size, pj, true);
 
         if (stage1_is_ok != simdjson::SUCCESS) {
             pj.error_code = stage1_is_ok;
             return pj.error_code;
         }
-
+#endif
         load_next_batch = false;
 
         //If we loaded a perfect amount of documents last time, we need to skip the first element,
@@ -89,7 +120,9 @@ int JsonStream::json_parse(ParsedJson &pj) {
             load_next_batch = true;
         }
 
-        else current_buffer_loc = pj.structural_indexes[next_json];
+        else {
+            current_buffer_loc = pj.structural_indexes[next_json];
+        }
     }
     //TODO: have a more precise error check
     //Give it two chances for now.  We assume the error is because the json was not loaded completely in this batch.
@@ -99,10 +132,51 @@ int JsonStream::json_parse(ParsedJson &pj) {
         error_on_last_attempt = true;
         res = json_parse(pj);
     }
-
     return res;
 }
 
+#ifdef SIMDJSON_THREADS_ENABLED
+size_t JsonStream::find_last_json(const ParsedJson &pj) {
+    auto last_i = pj.n_structural_indexes - 1;
+    if (pj.structural_indexes[last_i] == _batch_size)
+        last_i = pj.n_structural_indexes - 2;
+    auto arr_cnt = 0;
+    auto obj_cnt = 0;
+    for (auto i = last_i; i > 0; i--) {
+        auto idxb = pj.structural_indexes[i];
+        switch (_buf[idxb]) {
+            case ':':
+            case ',':
+                continue;
+            case '}':
+                obj_cnt--;
+                continue;
+            case ']':
+                arr_cnt--;
+                continue;
+            case '{':
+                obj_cnt++;
+                break;
+            case '[':
+                arr_cnt++;
+                break;
+        }
+        auto idxa = pj.structural_indexes[i - 1];
+        switch (_buf[idxa]) {
+            case '{':
+            case '[':
+            case ':':
+            case ',':
+                continue;
+        }
+        if (!arr_cnt && !obj_cnt)
+            return pj.structural_indexes[last_i+1];
+        return idxb;
+    }
+    return 0;
+}
+#endif
+
 size_t JsonStream::get_current_buffer_loc() const {
     return current_buffer_loc;
 }
@@ -115,18 +189,16 @@ size_t JsonStream::get_n_bytes_parsed() const {
     return n_bytes_parsed;
 }
 
-
 //// TODO: generalize this set of functions.  We don't want to have a copy in jsonparser.cpp
 void find_the_best_supported_implementation() {
+    uint32_t supports = detect_supported_architectures();
+    // Order from best to worst (within architecture)
+#ifdef IS_X86_64
     constexpr uint32_t haswell_flags =
             instruction_set::AVX2 | instruction_set::PCLMULQDQ |
             instruction_set::BMI1 | instruction_set::BMI2;
     constexpr uint32_t westmere_flags =
             instruction_set::SSE42 | instruction_set::PCLMULQDQ;
-
-    uint32_t supports = detect_supported_architectures();
-    // Order from best to worst (within architecture)
-#ifdef IS_X86_64
     if ((haswell_flags & supports) == haswell_flags) {
         best_stage1 = simdjson::find_structural_bits<Architecture ::HASWELL>;
         best_stage2 = simdjson::unified_machine<Architecture ::HASWELL>;

tests/CMakeLists.txt

@@ -10,6 +10,7 @@ add_cpp_test(jsonstream_test)
 add_cpp_test(pointercheck)
 add_cpp_test(integer_tests)
 
+target_link_libraries(jsonstream_test Threads::Threads)
 ## This causes problems
 # add_executable(singleheader ./singleheadertest.cpp ${PROJECT_SOURCE_DIR}/singleheader/simdjson.cpp)
 # target_compile_definitions(singleheader PRIVATE JSON_TEST_PATH="${PROJECT_SOURCE_DIR}/jsonexamples/twitter.json")