libdistributed version 0.0.3

Bug Fixes:

+ Prefix names with libdistributed to avoid name conflicts
+ Fix CMake Export to make it easier to use from other packages.
+ Fix invalid MPI_Isend call by taking the address of an object rather
  than passing it itself
+ Fix multiple send to actually send multiple values

Author: robertu94

CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.12)
-project(libdistributed VERSION "0.0.2" LANGUAGES CXX)
+project(libdistributed VERSION "0.0.3" LANGUAGES CXX)
 
 #correct was to set a default build type
 # https://blog.kitware.com/cmake-and-the-default-build-type/
@@ -31,8 +31,12 @@ add_library(libdistributed
     src/work_queue.cc
 
   #public headers
-    include/work_queue.h
-    include/types.h
+    include/libdistributed_stop_token.h
+    include/libdistributed_types.h
+    include/libdistributed_work_queue.h
+
+  #private headers
+    include/libdistributed_work_queue_impl.h
   )
 target_include_directories(
   libdistributed
@@ -50,13 +54,14 @@ if(USE_CLANG_TIDY)
   set_target_properties(libdistributed PROPERTIES C_CLANG_TIDY "${CLANG_TIDY}")
 endif()
 
-export(TARGETS libdistributed NAMESPACE LibDistributed:: FILE LibDistributed.cmake)
-install(TARGETS libdistributed EXPORT LibDistributed
+export(TARGETS libdistributed NAMESPACE LibDistributed:: FILE
+  LibDistributedConfig.cmake)
+install(TARGETS libdistributed EXPORT LibDistributedConfig
   ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
   LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
   RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
   )
-install(EXPORT LibDistributed NAMESPACE LibDistributed:: DESTINATION ${CMAKE_INSTALL_DATAROOTDIR}/LibDistributed/cmake)
+install(EXPORT LibDistributedConfig NAMESPACE LibDistributed:: DESTINATION ${CMAKE_INSTALL_DATAROOTDIR}/LibDistributed/cmake)
 install(DIRECTORY include/ DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/libdistributed)
 
 option(BUILD_DOCS "build the documetation" OFF)

include/stop_token.h include/libdistributed_stop_token.h

@@ -1,4 +1,5 @@
-#pragma once
+#ifndef LIBDISTRIBUTED_STOP_TOKEN_H
+#define LIBDISTRIBUTED_STOP_TOKEN_H
 
 /**
  * \file
@@ -32,3 +33,5 @@ class StopToken {
 
 }
 }
+
+#endif

include/types.h include/libdistributed_types.h

@@ -1,3 +1,5 @@
+#ifndef LIBDISTRIBUTED_TYPES_H
+#define LIBDISTRIBUTED_TYPES_H
 #include <array>
 #include <vector>
 #include <algorithm>
@@ -184,3 +186,5 @@ namespace distributed {
 
   }
 }
+
+#endif

include/work_queue.h include/libdistributed_work_queue.h

@@ -1,8 +1,10 @@
+#ifndef LIBDISTRIBUTED_WORK_QUEUE_H
+#define LIBDISTRIBUTED_WORK_QUEUE_H
 #include <mpi.h>
 #include <utility>
-#include "types.h"
-#include "stop_token.h"
-#include "work_queue_impl.h"
+#include "libdistributed_types.h"
+#include "libdistributed_stop_token.h"
+#include "libdistributed_work_queue_impl.h"
 
 /**
  * \file
@@ -88,3 +90,4 @@ void work_queue (
 
 }
 }
+#endif

include/work_queue_impl.h include/libdistributed_work_queue_impl.h

@@ -3,7 +3,7 @@
 #include <type_traits>
 #include <mpi.h>
 
-#include "stop_token.h"
+#include "libdistributed_stop_token.h"
 
 namespace distributed {
 namespace queue {
@@ -175,11 +175,12 @@ worker_send(MPI_Comm comm, MPI_Datatype response_dtype, IterableType iterable)
   auto value = *begin;
   MPI_Request mpi_request;
   for (; begin != end; ++begin) {
+    value = *begin;
     MPI_Isend(&value, 1, response_dtype, ROOT, (int)worker_status::more, comm,
               &mpi_request);
     MPI_Wait(&mpi_request, MPI_STATUS_IGNORE);
   }
-  MPI_Isend(value, 1, response_dtype, ROOT, (int)worker_status::done, comm, &mpi_request);
+  MPI_Isend(&value, 1, response_dtype, ROOT, (int)worker_status::done, comm, &mpi_request);
   MPI_Wait(&mpi_request, MPI_STATUS_IGNORE);
 }
 

test/mpi_test_main.cc

@@ -1,33 +1,6 @@
-#include <iostream>
-#include <string>
 #include "mpi.h"
-#include <unistd.h>
 #include "gtest/gtest.h"
 
-#include <execinfo.h>
-
-void failing_error_handler(MPI_Comm* comm, int* ec, ...) {
-  
-  int rank, size;
-  MPI_Comm_size(*comm, &size);
-  MPI_Comm_rank(*comm, &rank);
-  int nptrs;
-  void* backtrace_buffer[100];
-  char** backtrace_strings;
-  nptrs = backtrace(backtrace_buffer, 100);
-  backtrace_strings = backtrace_symbols(backtrace_buffer, nptrs);
-
-  for (int i = 0; i < nptrs; ++i) {
-    printf("BT %d/%d: [%d] %s\n", rank, size, i, backtrace_strings[i]);
-  }
-  free(backtrace_strings);
-
-
-  int length;
-  std::string s(MPI_MAX_ERROR_STRING, '\0');
-  MPI_Error_string(*ec, &s[0], &length);
-  ADD_FAILURE() << s.c_str();
-}
 
 int main(int argc, char *argv[])
 {

test/simple_queue.cc

@@ -5,7 +5,7 @@
 #include <cmath>
 
 #include <mpi.h>
-#include <work_queue.h>
+#include <libdistributed_work_queue.h>
 
 using namespace std::literals::chrono_literals;
 namespace queue = distributed::queue;

test/test_queue.cc

@@ -6,7 +6,7 @@
 #include <chrono>
 #include "gtest/gtest.h"
 
-#include "work_queue.h"
+#include "libdistributed_work_queue.h"
 
 using namespace distributed::queue;
 using namespace std::literals::chrono_literals;
@@ -44,6 +44,46 @@ TEST(test_work_queue, single_no_stop) {
   }
 }
 
+
+TEST(test_work_queue, multi_no_stop) {
+  using request = std::tuple<int>;
+  using response = std::tuple<int, double>;
+  std::vector<request> tasks;
+  tasks.reserve(5);
+  for (int i = 0; i < 5; ++i) {
+    tasks.emplace_back(i);
+  }
+  std::vector<response> results;
+
+  work_queue(
+      MPI_COMM_WORLD,
+      std::begin(tasks),
+      std::end(tasks),
+      [](request req) {
+        std::vector<response> responses;
+        responses.reserve(5);
+        int rank;
+        MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+        auto req_value = std::get<0>(req);
+        for (int i = 0; i < 5; ++i) {
+          responses.emplace_back(req_value, std::pow(req_value, 2));
+        }
+        return responses;
+      },
+      [&](response res) {
+        auto [i,d] = res;
+        results.push_back(res);
+      }
+      );
+
+  int rank;
+  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+  if(rank == 0) {
+    EXPECT_EQ(results.size(), 25);
+  } else {
+    EXPECT_EQ(results.size(), 0);
+  }
+}
 namespace std{
 inline void PrintTo(const std::chrono::milliseconds& duration, ::std::ostream * os) {
   *os << duration.count() << "ms";

test/test_types.cc

@@ -1,7 +1,7 @@
 #include <tuple>
 #include <vector>
 #include "gtest/gtest.h"
-#include "types.h"
+#include "libdistributed_types.h"
 
 using namespace distributed::types;