DSDE sample (n-body): added alternative approach (RMA accumulate) for metadata exchange

dynamic_sparse_data_exchange/configuration.c

@@ -9,7 +9,8 @@ const char *verbosity_levels[] = {"OFF", "INFO", "DEBUG", "TRACE"};
 
 const char *transmission_modes[] = {
    "Dynamic - MPI_Issend / MPI_Iprobe / MPI_Ibarrier",
-   "Collective - MPI_Alltoall / MPI_Alltoallw"
+   "Collective - MPI_Alltoall / MPI_Alltoallw",
+   "Remote Memory Access (RMA) - MPI_Accumulate / MPI_Probe"
 };
 
 // --------------------------------------------- Helper function declarations
@@ -55,6 +56,7 @@ const char* conf_set_from_args(conf_t *c, int argc, char* argv[], int nprocs)
       {"use-cart-topo", no_argument, &c->use_cartesian_topology, 1},
       {"collective", no_argument, &c->transmission_mode, COLLECTIVE},
       {"dynamic", no_argument, &c->transmission_mode, DYNAMIC},
+      {"rma", no_argument, &c->transmission_mode, RMA},
 
       {"ncells-x", required_argument, NULL, 'x'},
       {"ncells-y", required_argument, NULL, 'y'},

dynamic_sparse_data_exchange/configuration.h

@@ -14,7 +14,8 @@ enum verbosity_level_enum
 enum transmission_mode_enum
 {
    DYNAMIC = 0,
-   COLLECTIVE = 1
+   COLLECTIVE = 1,
+   RMA = 2
 };
 
 typedef struct

dynamic_sparse_data_exchange/mpicomm.c

@@ -8,7 +8,7 @@
 // --------------------------------------------- Helper function declarations
 
 int prepare_receive(part_t *particles, int n_particles_to_receive);
-void receive_particles(sim_t *s, const MPI_Status *status);
+int receive_particles(sim_t *s, const MPI_Status *status);
 
 void sim_info_reduce_op(void *invec, void *inoutvec, int *len, MPI_Datatype *type);
 
@@ -135,7 +135,69 @@ void mpi_communicate_particles_dynamic(sim_t *s)
 #endif
 }
 
-void receive_particles(sim_t *s, const MPI_Status *status)
+void mpi_communicate_particles_rma(sim_t *s)
+{
+   // Create MPI datatypes for sending particle data
+   MPI_Datatype *send_types;
+   MPI_Request *send_requests;
+   int *send_counts, *receiver_ranks;
+   int n_receivers;
+
+   mpitype_part_init_send(&s->leaving_particles, &send_types, &send_counts, &receiver_ranks, &n_receivers);
+
+   // Let other processes increase local recv_count (using remote memory access)
+   static MPI_Win recv_count_window = MPI_WIN_NULL;
+   static int recv_count;
+   int i, tag = 0;
+
+   if(recv_count_window == MPI_WIN_NULL) {
+      MPI_Aint window_size = sizeof(recv_count);
+      int displacement_unit = sizeof(int);
+
+      MPI_Win_create(
+            &recv_count, window_size, displacement_unit,
+            MPI_INFO_NULL, s->communicator,
+            &recv_count_window
+      );
+   }
+   recv_count = 0;
+   MPI_Win_fence(0, recv_count_window);
+   for(i = 0; i < n_receivers; i++) {
+      MPI_Aint target_displacement = 0;
+      MPI_Accumulate(
+            &send_counts[i], 1, MPI_INT,     // send buffer
+            receiver_ranks[i],               // target rank
+            target_displacement, 1, MPI_INT, // target buffer
+            MPI_SUM, recv_count_window       // operation and window object
+      );
+   }
+   MPI_Win_fence(0, recv_count_window);
+
+   // Post non-blocking sends
+   send_requests = malloc(n_receivers * sizeof(*send_requests));
+   for(i = 0; i < n_receivers; i++) {
+      MPI_Isend(
+            MPI_BOTTOM, 1, send_types[i], receiver_ranks[i],
+            tag, s->communicator, &send_requests[i]
+      );
+   }
+
+   // Receive particles
+   MPI_Status status;
+   while(recv_count > 0) {
+      MPI_Probe(MPI_ANY_SOURCE, tag, s->communicator, &status);
+      recv_count -= receive_particles(s, &status);
+   }
+
+   // Finish sends
+   MPI_Waitall(n_receivers, send_requests, MPI_STATUSES_IGNORE);
+
+   // Cleanup
+   mpitype_part_free_send(&send_types, &send_counts, &receiver_ranks, &n_receivers);
+   free(send_requests);
+}
+
+int receive_particles(sim_t *s, const MPI_Status *status)
 {
    part_t *particles = &s->local_particles;
 
@@ -155,6 +217,8 @@ void receive_particles(sim_t *s, const MPI_Status *status)
 
    // Free MPI Datatype
    mpitype_part_free_recv(&recv_type);
+
+   return recv_count;
 }
 
 void mpi_communicate_particles_collective(sim_t *s)

dynamic_sparse_data_exchange/mpicomm.h

@@ -10,6 +10,8 @@ void mpi_reduce_sim_info(const sim_info_t *local_info, sim_info_t *info);
 
 void mpi_communicate_particles_dynamic(sim_t *s);
 
+void mpi_communicate_particles_rma(sim_t *s);
+
 void mpi_communicate_particles_collective(sim_t *s);
 
 #endif

dynamic_sparse_data_exchange/simulation.c

@@ -116,6 +116,8 @@ void sim_communicate_particles(sim_t *s)
           mpi_communicate_particles_dynamic(s); break;
       case COLLECTIVE:
           mpi_communicate_particles_collective(s); break;
+      case RMA:
+          mpi_communicate_particles_rma(s); break;
    }
 }