Merge pull request #58 from DARMA-tasking/18-read-json-concurrently

#18: Read json concurrently

CMakeLists.txt

@@ -25,9 +25,19 @@ endif()
 message(STATUS "CMAKE_CXX_STANDARD: ${CMAKE_CXX_STANDARD}")
 
 option(vt_tv_python_bindings_enabled "Build vt-tv with Python bindings" ON)
+option(vt_tv_openmp_enabled "Build vt-tv with openMP support" ON)
 
 include(cmake/load_packages.cmake)
 
+if(APPLE)
+  add_compile_options(-ffat-lto-objects)
+endif()
+
+if(openmp_enabled)
+  set(VT_TV_NUM_THREADS "2" CACHE STRING "Number of threads to use")
+  add_definitions(-DVT_TV_NUM_THREADS=${VT_TV_NUM_THREADS})
+endif()
+
 add_custom_target(vt_tv_examples)
 add_custom_target(vt_tv_tests)
 add_custom_target(vt_tv_apps)

cmake/load_openmp.cmake

@@ -0,0 +1,10 @@
+if(NOT APPLE)
+  find_package(OpenMP)
+  if (OPENMP_FOUND)
+    message(STATUS "Found OpenMP")
+    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${OpenMP_C_FLAGS}")
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${OpenMP_CXX_FLAGS}")
+    set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} ${OpenMP_EXE_LINKER_FLAGS}")
+    set(openmp_enabled ON)
+  endif()
+endif()

cmake/load_packages.cmake

@@ -31,3 +31,7 @@ include(cmake/load_vtk_package.cmake)
 if (vt_tv_python_bindings_enabled)
   include(cmake/load_nanobind_package.cmake)
 endif()
+
+if (vt_tv_openmp_enabled)
+  include(cmake/load_openmp.cmake)
+endif()

src/vt-tv/api/info.h

@@ -437,6 +437,6 @@ struct Info {
   std::unordered_map<NodeType, Rank> ranks_;
 };
 
-} /* end namesapce vt::tv */
+} /* end namespace vt::tv */
 
 #endif /*INCLUDED_VT_TV_API_INFO_H*/

src/vt-tv/api/object_communicator.h

@@ -213,6 +213,6 @@ struct ObjectCommunicator {
   std::multimap<ElementIDType, double> sent_;     /**< The sent edges */
 };
 
-} /* end namesapce vt::tv */
+} /* end namespace vt::tv */
 
 #endif /*INCLUDED_VT_TV_API_OBJECT_COMMUNICATOR_H*/

src/vt-tv/api/object_info.h

@@ -188,6 +188,6 @@ struct ObjectInfo {
   bool is_collection_ = false;
 };
 
-} /* end namesapce vt::tv */
+} /* end namespace vt::tv */
 
 #endif /*INCLUDED_VT_TV_API_OBJECT_INFO_H*/

src/vt-tv/api/object_work.h

@@ -190,6 +190,6 @@ struct ObjectWork {
   ObjectCommunicator communicator_;
 };
 
-} /* end namesapce vt::tv */
+} /* end namespace vt::tv */
 
 #endif /*INCLUDED_VT_TV_API_OBJECT_WORK_H*/

src/vt-tv/api/phase_work.h

@@ -138,6 +138,6 @@ struct PhaseWork {
   std::unordered_map<ElementIDType, ObjectWork> objects_;
 };
 
-} /* end namesapce vt::tv */
+} /* end namespace vt::tv */
 
 #endif /*INCLUDED_VT_TV_API_PHASE_WORK_H*/

src/vt-tv/api/rank.h

@@ -116,6 +116,6 @@ struct Rank {
   std::unordered_map<PhaseType, PhaseWork> phase_info_;
 };
 
-} /* end namesapce vt::tv */
+} /* end namespace vt::tv */
 
 #endif /*INCLUDED_VT_TV_API_RANK_H*/

src/vt-tv/api/types.h

@@ -56,6 +56,6 @@ using UniqueIndexBitType = uint64_t;
 using TimeType = double;
 using CollectionObjGroupIDType = uint64_t;
 
-} /* end namesapce vt::tv */
+} /* end namespace vt::tv */
 
 #endif /*INCLUDED_VT_TV_API_TYPES_H*/

src/vt-tv/render/render.cc

@@ -97,7 +97,7 @@ Render::Render(Info in_info)
 Render::Render(
   std::array<std::string, 3> in_qoi_request,
   bool in_continuous_object_qoi,
-  Info in_info,
+  Info& in_info,
   std::array<uint64_t, 3> in_grid_size,
   double in_object_jitter,
   std::string in_output_dir,
@@ -348,7 +348,6 @@ vtkNew<vtkPolyData> Render::createObjectMesh_(PhaseType phase) {
   points->SetNumberOfPoints(n_o);
 
   // Retrieve elements constant across all ranks
-  std::vector<NodeType> ranks = this->info_.getRankIDs();
   std::string object_qoi = this->object_qoi_;
 
   // Iterate over ranks and objects to create mesh points
@@ -938,14 +937,13 @@ void Render::renderPNG(
 }
 
 void Render::generate(uint64_t font_size, uint64_t win_size) {
-  std::pair<double, double> rank_qoi_range = this->computeRankQoiRange_();
-  double rank_qoi_min = std::get<0>(rank_qoi_range);
-  double rank_qoi_max = std::get<1>(rank_qoi_range);
+  double rank_qoi_min = rank_qoi_range_.first;
+  double rank_qoi_max = rank_qoi_range_.second;
 
   if (std::holds_alternative<std::pair<double, double>>(object_qoi_range_)) {
     auto range_pair = std::get<std::pair<double, double>>(object_qoi_range_);
-    double object_qoi_min = std::get<0>(range_pair);
-    double object_qoi_max = std::get<1>(range_pair);
+    double object_qoi_min = range_pair.first;
+    double object_qoi_max = range_pair.second;
     fmt::print("Rank {} range: {}, {}\n", rank_qoi_, rank_qoi_min, rank_qoi_max);
     fmt::print("Object {} range: {}, {}\n", object_qoi_, object_qoi_min, object_qoi_max);
   }
@@ -990,7 +988,6 @@ void Render::generate(uint64_t font_size, uint64_t win_size) {
           std::cerr << e.what() << '\n';
           obj_qoi_range = {0, 1};
         }
-        auto load_range = this->computeRankQoiRange_();
 
         uint64_t window_size = win_size;
         uint64_t edge_width = 0.03 * window_size / *std::max_element(this->grid_size_.begin(), this->grid_size_.end());

src/vt-tv/render/render.h

@@ -264,7 +264,7 @@ struct Render {
   Render(
     std::array<std::string, 3> in_qoi_request,
     bool in_continuous_object_qoi,
-    Info in_info,
+    Info& in_info,
     std::array<uint64_t, 3> in_grid_size,
     double in_object_jitter,
     std::string in_output_dir,
@@ -304,6 +304,6 @@ struct Render {
   void generate(uint64_t font_size = 50, uint64_t win_size = 2000);
 };
 
-}} /* end namesapce vt::tv */
+}} /* end namespace vt::tv */
 
 #endif /*INCLUDED_VT_TV_RENDER_RENDER_H*/

src/vt-tv/utility/json_generator.cc

@@ -118,4 +118,4 @@ void JSONGenerator::outputObjectMetaData(nlohmann::json& j, ElementIDType id) co
   }
 }
 
-} /* end namesapce vt::tv::utility */
+} /* end namespace vt::tv::utility */

src/vt-tv/utility/json_generator.h

@@ -93,6 +93,6 @@ struct JSONGenerator {
   PhaseType phase_ = 0;
 };
 
-} /* end namesapce vt::tv::utility */
+} /* end namespace vt::tv::utility */
 
 #endif /*INCLUDED_VT_TV_UTILITY_JSON_GENERATOR_H*/

src/vt-tv/utility/json_reader.cc

@@ -217,10 +217,14 @@ std::unique_ptr<Info> JSONReader::parseFile() {
             // fmt::print(" From: {}, to: {}\n", from_id, to_id);
 
             // Object on this rank sent data
-            if (objects.find(from_id) != objects.end()) {
-              objects.at(from_id).addSentCommunications(to_id, bytes);
-            } else if (objects.find(to_id) != objects.end()) {
-              objects.at(to_id).addReceivedCommunications(from_id, bytes);
+            auto from_it = objects.find(from_id);
+            if (from_it != objects.end()) {
+              from_it->second.addSentCommunications(to_id, bytes);
+            } else {
+              auto to_it = objects.find(to_id);
+              if (to_it != objects.end()) {
+                to_it->second.addReceivedCommunications(from_id, bytes);
+              }
             }
           }
         }
@@ -237,4 +241,4 @@ std::unique_ptr<Info> JSONReader::parseFile() {
   return std::make_unique<Info>(std::move(object_info), std::move(rank_info));
 }
 
-} /* end namesapce vt::tv::utility */
+} /* end namespace vt::tv::utility */

src/vt-tv/utility/json_reader.h

@@ -86,8 +86,6 @@ struct JSONReader {
   /**
    * \brief Parse the json into vt-tv's data structure Info, with a single rank
    * filled out
-   *
-   * \return vt-tv Info
    */
   std::unique_ptr<Info> parseFile();
 
@@ -97,6 +95,6 @@ struct JSONReader {
   std::unique_ptr<nlohmann::json> json_ = nullptr;
 };
 
-} /* end namesapce vt::tv::utility */
+} /* end namespace vt::tv::utility */
 
 #endif /*INCLUDED_VT_TV_UTILITY_JSON_READER_H*/

src/vt-tv/utility/parse_render.cc

@@ -72,19 +72,30 @@ void ParseRender::parseAndRender(PhaseType phase_id, std::unique_ptr<Info> info)
         input_dir += '/';
       }
 
-      uint64_t n_ranks = config["input"]["n_ranks"].as<uint64_t>();
+      int64_t n_ranks = config["input"]["n_ranks"].as<int64_t>(); // signed for omp parallel for
 
       // Read JSON file and input data
       std::filesystem::path p = input_dir;
       std::string path = std::filesystem::absolute(p).string();
 
       info = std::make_unique<Info>();
 
-      for (NodeType rank = 0; rank < n_ranks; rank++) {
-        utility::JSONReader reader{rank, input_dir + "data." + std::to_string(rank) + ".json"};
+      #ifdef VT_TV_NUM_THREADS
+        const int threads = VT_TV_NUM_THREADS;
+      #else
+        const int threads = 2;
+      #endif
+      omp_set_num_threads(threads);
+      # pragma omp parallel for
+      for (int64_t rank = 0; rank < n_ranks; rank++) {
+        fmt::print("Reading file for rank {}\n", rank);
+        utility::JSONReader reader{static_cast<NodeType>(rank), input_dir + "data." + std::to_string(rank) + ".json"};
         reader.readFile();
         auto tmpInfo = reader.parseFile();
+        #pragma omp critical
+        {
         info->addInfo(tmpInfo->getObjectInfo(), tmpInfo->getRank(rank));
+        }
       }
     }
 
@@ -139,7 +150,7 @@ void ParseRender::parseAndRender(PhaseType phase_id, std::unique_ptr<Info> info)
 
     // Instantiate render
     Render r(
-      qoi_request, continuous_object_qoi, *info, grid_size, object_jitter,
+      qoi_request, continuous_object_qoi, *std::move(info), grid_size, object_jitter,
       output_dir, output_file_stem, 1.0, save_meshes, save_pngs, phase_id
     );
     r.generate(font_size, win_size);
@@ -149,4 +160,4 @@ void ParseRender::parseAndRender(PhaseType phase_id, std::unique_ptr<Info> info)
   }
 }
 
-} /* end namesapce vt::tv::utility */
+} /* end namespace vt::tv::utility */

src/vt-tv/utility/parse_render.h

@@ -51,6 +51,8 @@
 #include <limits>
 #include <memory>
 
+#include <omp.h>
+
 namespace vt::tv::utility {
 
 /**
@@ -86,6 +88,6 @@ struct ParseRender {
   std::string filename_;
 };
 
-} /* end namesapce vt::tv::utility */
+} /* end namespace vt::tv::utility */
 
 #endif /*INCLUDED_VT_TV_UTILITY_PARSE_RENDER_H*/