LibDistributed version 0.0.9

Major Changes:

+ Added serialization support for std::string
+ Added work_queue_options to provide customization points for
  parallelism, specifically work group sizes and root process
+ Queue can now have multiple processes per worker or for the master
+ TaskManager now returns get_subcommunicator for nested parallelism

Minor Changes:

Encapsulated some of the calls to MPI, future versions of LibDistributed
_may_ support different distributed programming frameworks

Author: robertu94

include/libdistributed_comm.h

@@ -10,6 +10,7 @@
 #include <variant>
 #include <vector>
 #include <map>
+#include <string>
 #include <mpi.h>
 
 /**
@@ -730,6 +731,91 @@ struct serializer<std::vector<T>>
   }
 };
 
+/** specialization for serialization for vector types */
+template <class CharT>
+struct serializer<std::basic_string<CharT>>
+{
+  /** is the type serializable using MPI_Datatypes for both the sender and
+   * receiver at compile time?*/
+  using mpi_type = std::false_type;
+  /** \returns a string representing the name of the type */
+  static std::string name() { 
+    std::stringstream n;
+    n << "std::string<" << serializer<CharT>::name() << ">";
+    return n.str();
+  }
+  /** \returns a MPI_Datatype to represent the type if mpi_type is true, else MPI_INT */
+  static MPI_Datatype dtype() { return MPI_INT; }
+  /** 
+   * Sends a data type from one location to another
+   * \param[in] t the data to send
+   * \param[in] dest the MPI rank to send to
+   * \param[in] tag the MPI tag to send to
+   * \param[in] comm the MPI_Comm to send to
+   * \returns an error code from the underlying send */
+  static int send(std::basic_string<CharT> const& t, int dest, int tag, MPI_Comm comm)
+  {
+    int ret = 0;
+    size_t size = t.size();
+    ret = MPI_Send(&size, 1, mpi_size_t(), dest, tag, comm);
+    if(serializer<CharT>::mpi_type::value) {
+      return MPI_Send(&t.front(), t.size(), serializer<CharT>::dtype(), dest, tag, comm);
+    } else { 
+      for (auto const& item : t) {
+        ret |= serializer<CharT>::send(item, dest, tag, comm);
+      }
+    }
+    return ret;
+  }
+  /** 
+   * Recv a data type from another location
+   * \param[in] t the data to recv from
+   * \param[in] source the MPI rank to recv from
+   * \param[in] tag the MPI tag to recv from
+   * \param[in] comm the MPI_Comm to recv from
+   * \param[in] status the MPI_Status to recv from
+   * \returns an error code from the underlying recv */
+  static int recv(std::basic_string<CharT>& t, int source, int tag, MPI_Comm comm,
+                  MPI_Status* status)
+  {
+    MPI_Status alt_status;
+    if(status == MPI_STATUS_IGNORE) status = &alt_status;
+    size_t size, ret=0;
+    ret = MPI_Recv(&size, 1, mpi_size_t(), source, tag, comm, status);
+    t.resize(size);
+
+    if (serializer<CharT>::mpi_type::value) {
+      return MPI_Recv(&t.front(), t.size(), serializer<CharT>::dtype(), status->MPI_SOURCE, status->MPI_TAG, comm, status);
+    } else {
+      for (auto& item : t) {
+        ret |= serializer<CharT>::recv(item, status->MPI_SOURCE, status->MPI_TAG, comm, status);
+      }
+    }
+    return ret;
+  }
+
+  /** 
+   * Broadcast a data type from another location
+   * \param[in] t the data to broadcast from
+   * \param[in] root the MPI rank to broadcast from
+   * \param[in] comm the MPI_Comm to broadcast from
+   * \returns an error code from the underlying MPI_Bcast(s) */
+  static int bcast(std::basic_string<CharT>& t, int root, MPI_Comm comm)
+  {
+    size_t size = t.size(), ret=0;
+    ret = MPI_Bcast(&size, 1, mpi_size_t(), root, comm);
+    t.resize(size);
+    if (serializer<CharT>::mpi_type::value) {
+      return MPI_Bcast(&t.front(), t.size(), serializer<CharT>::dtype(), root, comm);
+    } else {
+      for (auto& item : t) {
+        ret |= serializer<CharT>::bcast(item, root, comm);
+      }
+    }
+    return ret;
+  }
+};
+
 template <class T, size_t N>
 struct serializer<std::array<T, N>>
 {

include/libdistributed_task_manager.h

@@ -32,7 +32,7 @@ class StopToken {
  * A distributed token which can be used to request stopping computation on a
  * work_queue
  */
-template <class RequestType>
+template <class RequestType, class CommunicatorType>
 class TaskManager: public StopToken {
   public:
   virtual ~TaskManager()=default;
@@ -42,6 +42,11 @@ class TaskManager: public StopToken {
    * \param[in] request the request that you would like the master to enqueue
    */
   virtual void push(RequestType const& request)=0;
+
+  /**
+   * Request a sub-communicator for the current process group
+   */
+  virtual CommunicatorType get_subcommunicator()=0;
 };
 
 }

include/libdistributed_work_queue.h

@@ -4,6 +4,7 @@
 #include <utility>
 #include "libdistributed_types.h"
 #include "libdistributed_task_manager.h"
+#include "libdistributed_work_queue_options.h"
 #include "libdistributed_work_queue_impl.h"
 
 /**
@@ -44,41 +45,58 @@ namespace queue {
    *
    * \see distributed::queue::TaskManager<RequestType> for details on the semantics about cancellation
    */
-template <class TaskForwardIt, class WorkerFunction, class MasterFunction>
+template <class TaskRandomIt, class WorkerFunction, class MasterFunction>
 void work_queue (
-    MPI_Comm comm,
-    TaskForwardIt tasks_begin,
-    TaskForwardIt tasks_end,
+    work_queue_options<typename impl::iterator_to_value_type<TaskRandomIt>::type> const& options,
+    TaskRandomIt tasks_begin,
+    TaskRandomIt tasks_end,
     WorkerFunction worker_fn,
     MasterFunction master_fn
     ) {
   //setup communicator
-  int rank, size;
-  MPI_Comm queue_comm;
-  MPI_Comm_dup(comm, &queue_comm);
-  MPI_Comm_rank(queue_comm, &rank);
-  MPI_Comm_size(queue_comm, &size);
 
-  using RequestType = typename impl::iterator_to_value_type<TaskForwardIt>::type;
+  using RequestType = typename impl::iterator_to_value_type<TaskRandomIt>::type;
   using ResponseType = decltype( impl::maybe_stop_token( worker_fn,
         std::declval<RequestType>(),
-        std::declval<TaskManager<RequestType>&>()
+        std::declval<TaskManager<RequestType, MPI_Comm>&>()
         )
       );
+  
+
+  if(options.get_queue_size() > 1) {
+    //create sub-communicators
+    const int rank = options.get_queue_rank();
+    auto groups = options.get_groups();
+    MPI_Comm subcomm;
+    MPI_Comm_split(
+        options.get_native_queue_comm(),
+        groups[rank],
+        rank,
+        &subcomm
+        );
+    int subrank;
+    MPI_Comm_rank(subcomm, &subrank);
 
-  if(size > 1) {
     //determine the request and response types from the input
-    if(rank == 0) {
-      impl::master<RequestType, ResponseType>(queue_comm, tasks_begin, tasks_end, master_fn);
+    if(options.is_master()) {
+      if(subrank == 0) {
+        impl::master_main<RequestType, ResponseType>(subcomm, tasks_begin, tasks_end, master_fn, options);
+      } else {
+        impl::master_aux<RequestType, ResponseType>(subcomm, master_fn, options);
+      }
     } else {
-      impl::worker<RequestType, ResponseType>(queue_comm, worker_fn);
+      if(subrank == 0) {
+        impl::worker_main<RequestType, ResponseType>(subcomm, worker_fn, options);
+      } else {
+        impl::worker_aux<RequestType, ResponseType>(subcomm, worker_fn, options);
+      }
     }
+    MPI_Comm_free(&subcomm);
 
   } else {
     impl::no_workers<RequestType, ResponseType>(tasks_begin, tasks_end, master_fn, worker_fn);
   }
   comm::serializer::get_type_registry().clear();
-  MPI_Comm_free(&queue_comm);
 }
 
 }