Merge pull request #73 from LLNL/develop

v0.4 Prep

Author: steiltre

CMakeLists.txt

@@ -99,7 +99,8 @@ if (NOT cereal_FOUND)
         # Populate cereal
         FetchContent_Populate(cereal)
         # Include cereal root cmake boilerplate
-        add_subdirectory(${cereal_SOURCE_DIR})
+        add_subdirectory(${cereal_SOURCE_DIR} ${cereal_BINARY_DIR})
+
         message(STATUS "YGM cloned cereal dependency: " ${cereal_SOURCE_DIR})
     endif ()
     target_link_libraries(ygm INTERFACE cereal::cereal)

Readme.md

@@ -27,21 +27,29 @@ perform a user-provided function only on the desired data. These containers are
 YGM is a header-only library that is easy to incorporate into a project through CMake. Adding the following to
 CMakeLists.txt will install YGM and its dependencies as part of your project:
 ```
-find_package(ygm CONFIG)
-if(NOT ygm_FOUND)
+set(DESIRED_YGM_VERSION 0.3)
+find_package(ygm ${DESIRED_YGM_VERSION} CONFIG)
+if (NOT ygm_FOUND)
     FetchContent_Declare(
         ygm
         GIT_REPOSITORY https://github.com/LLNL/ygm
-        GIT_TAG        <commit hash here>         
-    )         
-    set(JUST_INSTALL_YGM ON)
-    set(YGM_INSTALL ON)
-    FetchContent_MakeAvailable(ygm)
-    message(STATUS "Cloned ygm dependency " ${ygm_SOURCE_DIR})
-else()
-    message(STATUS "Found ygm dependency " ${ygm_DIR})
-endif()
-target_link_libraries(my_cool_project INTERFACE ygm::ygm)
+        GIT_TAG v${DESIRED_YGM_VERSION}
+    )
+    FetchContent_GetProperties(ygm)
+    if (ygm_POPULATED)
+        message(STATUS "Found already populated ygm dependency: "
+                       ${ygm_SOURCE_DIR}
+        )
+    else ()
+        set(JUST_INSTALL_YGM ON)
+        set(YGM_INSTALL ON)
+        FetchContent_Populate(ygm)
+        add_subdirectory(${ygm_SOURCE_DIR} ${ygm_BINARY_DIR})
+        message(STATUS "Cloned ygm dependency " ${ygm_SOURCE_DIR})
+    endif ()
+else ()
+    message(STATUS "Found installed ygm dependency " ${ygm_DIR})
+endif ()
 ```
 
 # Anatomy of a YGM Program

include/ygm/comm.hpp

@@ -14,13 +14,20 @@
 namespace ygm {
 
 class comm {
+ private:
+  class impl;
+
  public:
   comm(int *argc, char ***argv, int buffer_capacity);
 
   // TODO:  Add way to detect if this MPI_Comm is already open. E.g., static
   // map<MPI_Comm, impl*>
   comm(MPI_Comm comm, int buffer_capacity);
 
+  // Constructor to allow comm::impl to build temporary comm using itself as the
+  // impl
+  comm(std::shared_ptr<impl> impl_ptr);
+
   ~comm();
 
   //
@@ -150,7 +157,6 @@ class comm {
  private:
   comm() = delete;
 
-  class impl;
   std::shared_ptr<impl>                      pimpl;
   std::shared_ptr<detail::mpi_init_finalize> pimpl_if;
 };

include/ygm/detail/comm_impl.hpp

@@ -19,7 +19,7 @@
 
 namespace ygm {
 
-class comm::impl {
+class comm::impl : public std::enable_shared_from_this<comm::impl> {
  public:
   impl(MPI_Comm c, int buffer_capacity) {
     ASSERT_MPI(MPI_Comm_dup(c, &m_comm_async));
@@ -35,7 +35,6 @@ class comm::impl {
   }
 
   ~impl() {
-    barrier();
     // send kill signal to self (listener thread)
     ASSERT_RELEASE(MPI_Send(NULL, 0, MPI_BYTE, m_comm_rank, 0, m_comm_async) ==
                    MPI_SUCCESS);
@@ -399,12 +398,12 @@ class comm::impl {
         std::forward<const PackArgs>(args)...);
     ASSERT_DEBUG(sizeof(Lambda) == 1);
 
-    void (*fun_ptr)(impl *, cereal::YGMInputArchive &) =
-        [](impl *t, cereal::YGMInputArchive &bia) {
+    void (*fun_ptr)(comm *, cereal::YGMInputArchive &) =
+        [](comm *c, cereal::YGMInputArchive &bia) {
           std::tuple<PackArgs...> ta;
           bia(ta);
           Lambda *pl = nullptr;
-          auto    t1 = std::make_tuple((impl *)t);
+          auto    t1 = std::make_tuple((comm *)c);
 
           // \pp was: std::apply(*pl, std::tuple_cat(t1, ta));
           ygm::meta::apply_optional(*pl, std::move(t1), std::move(ta));
@@ -430,6 +429,7 @@ class comm::impl {
    */
   bool process_receive_queue() {
     bool received = false;
+    comm tmp_comm(shared_from_this());
     while (true) {
       auto buffer = receive_queue_try_pop();
       if (buffer.second == 0) break;
@@ -439,9 +439,9 @@ class comm::impl {
         int64_t iptr;
         iarchive(iptr);
         iptr += (int64_t)&reference;
-        void (*fun_ptr)(impl *, cereal::YGMInputArchive &);
+        void (*fun_ptr)(comm *, cereal::YGMInputArchive &);
         memcpy(&fun_ptr, &iptr, sizeof(uint64_t));
-        fun_ptr(this, iarchive);
+        fun_ptr(&tmp_comm, iarchive);
         m_recv_count++;
         m_local_rpc_calls++;
       }
@@ -496,10 +496,13 @@ inline comm::comm(MPI_Comm mcomm, int buffer_capacity = 16 * 1024 * 1024) {
   pimpl = std::make_shared<comm::impl>(mcomm, buffer_capacity);
 }
 
+inline comm::comm(std::shared_ptr<impl> impl_ptr) : pimpl(impl_ptr) {}
+
 inline comm::~comm() {
-  ASSERT_RELEASE(MPI_Barrier(MPI_COMM_WORLD) == MPI_SUCCESS);
+  if (pimpl.use_count() == 1) {
+    barrier();
+  }
   pimpl.reset();
-  ASSERT_RELEASE(MPI_Barrier(MPI_COMM_WORLD) == MPI_SUCCESS);
   pimpl_if.reset();
 }
 

include/ygm/io/csv_parser.hpp

@@ -28,15 +28,16 @@ class csv_parser {
   void for_all(Function fn) {
     using namespace ygm::io::detail;
     m_lp.for_all([fn](const std::string& line) {
-      auto vfields   = parse_csv_line(line);
-      //auto stypes    = convert_type_string(vfields);
-      //todo, detect if types are inconsistent between records
-      fn(vfields);
+      auto vfields = parse_csv_line(line);
+      // auto stypes    = convert_type_string(vfields);
+      // todo, detect if types are inconsistent between records
+      if (vfields.size() > 0) {
+        fn(vfields);
+      }
     });
   }
 
  private:
   line_parser m_lp;
 };
-
 }  // namespace ygm::io