support multi-thread for converter

Signed-off-by: Lei Wang <[email protected]>

.dockerignore

@@ -109,6 +109,8 @@ default.etcd/
 # outer build directory
 build/
 analytical_engine/build/
+converter/build/
+vegito/build/
 
 # pom generated by maven-flatten-plugin
 **/.flattened-pom.xml

.gitmodules

@@ -1,3 +1,6 @@
 [submodule "interfaces/grin/include"]
 	path = interfaces/grin/include
 	url = https://github.com/GraphScope/GRIN.git
+[submodule "third_party/abseil-cpp"]
+	path = third_party/abseil-cpp
+	url = https://github.com/abseil/abseil-cpp.git

charts/gart/templates/converter/deployment.yaml

@@ -81,6 +81,9 @@ spec:
           --use-logs-per-epoch ${USE_LOGS_PER_EPOCH} \
           --logs-per-epoch ${LOGS_PER_EPOCH} \
           --seconds-per-epoch ${SECONDS_PER_EPOCH} \
+          {{- if .Values.dataconfig.numThreadsForConverter }}
+          --num-threads-for-converter {{ .Values.dataconfig.numThreadsForConverter }} \
+          {{- end }}
           --rg-from-etcd true \
           --k8s-mode true \
           --role converter &&

charts/gart/values.yaml

@@ -253,3 +253,5 @@ dataconfig:
   # defaultMaxVertexNum: 67108864
   # defaultMaxMemUsage: 10737418240
   # customMemConfig: TITLE:67108861:1073741824,NAME:67108861:1073741824
+  # set the number of threads for converter, if unset, use the default value 4
+  # numThreadsForConverter: 4

converter/CMakeLists.txt

@@ -20,6 +20,17 @@ if(NOT CMAKE_BUILD_TYPE AND NOT CMAKE_CONFIGURATION_TYPES)
   set(CMAKE_BUILD_TYPE "Release" CACHE STRING "The default build type is Release" FORCE)
 endif()
 
+option (USE_TBB "use TBB for concurrent queue?" OFF)
+option (USE_FLAT_HASH_MAP "use absl::flat_hash_map?" ON)
+
+if(USE_TBB)
+  add_definitions(-DUSE_TBB)
+endif()
+
+if (USE_FLAT_HASH_MAP)
+  add_definitions(-DUSE_FLAT_HASH_MAP)
+endif()
+
 
 # ------------------------------------------------------------------------------
 # find_libraries
@@ -54,11 +65,27 @@ if (GLOG_FOUND)
     set(CMAKE_REQUIRED_LIBRARIES "${CMAKE_REQUIRED_LIBRARIES} ${GLOG_LIBRARIES}")
 endif ()
 
+if(USE_TBB)
+  # find TBB
+  include("../vegito/cmake/FindTBB.cmake")
+  if (TBB_FOUND)
+      include_directories(${TBB_INCLUDE_DIRS})
+      link_directories(${TBB_LIBRARY_DIRS})
+      link_libraries(${TBB_LIBRARIES})
+  else ()
+      message(SEND_ERROR "TBB not found")
+  endif ()
+endif ()
+
 if(ENABLE_CHECKPOINT)
-  find_package(Boost REQUIRED COMPONENTS serialization)
+  find_package(Boost REQUIRED COMPONENTS)
   include_directories(${Boost_INCLUDE_DIRS})
 endif()
 
+if (USE_FLAT_HASH_MAP)
+  add_subdirectory(../third_party/abseil-cpp abseil-cpp)
+endif()
+
 file(GLOB_RECURSE FILES_NEED_FORMAT "*.cc" "*.h")
 
 add_custom_target(convert_clformat
@@ -80,6 +107,10 @@ if(ENABLE_CHECKPOINT)
   target_link_libraries(binlog_convert_debezium ${Boost_LIBRARIES})
 endif()
 
+if (USE_FLAT_HASH_MAP)
+  target_link_libraries(binlog_convert_debezium absl::flat_hash_map)
+endif()
+
 add_executable(binlog_convert_maxwell binlog_convert.cc flags.cc parser.cc)
 
 target_include_directories(binlog_convert_maxwell PRIVATE ${RDKAFKA_INCLUDE_DIR})
@@ -92,3 +123,7 @@ if(ENABLE_CHECKPOINT)
   target_compile_definitions(binlog_convert_maxwell PRIVATE ENABLE_CHECKPOINT)
   target_link_libraries(binlog_convert_maxwell ${Boost_LIBRARIES})
 endif()
+
+if (USE_FLAT_HASH_MAP)
+  target_link_libraries(binlog_convert_maxwell absl::flat_hash_map)
+endif()