src/smpi/smpi_rma.c

   1
   2 /* Copyright (c) 2007-2014. The SimGrid Team.
   3  * All rights reserved.                                                     */
   4
   5 /* This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the license (GNU LGPL) which comes with this package. */
   7
   8 #include "private.h"
   9
  10 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_rma, smpi, "Logging specific to SMPI (RMA operations)");
  11
  12 #define RMA_TAG -1234
  13
  14 xbt_bar_t creation_bar = NULL;
  15
  16 typedef struct s_smpi_mpi_win{
  17   void* base;
  18   MPI_Aint size;
  19   int disp_unit;
  20   MPI_Comm comm;
  21   //MPI_Info info
  22   int assert;
  23   xbt_dynar_t requests;
  24   xbt_bar_t bar;
  25   MPI_Win* connected_wins;
  26   char* name;
  27 } s_smpi_mpi_win_t;
  28
  29
  30 MPI_Win smpi_mpi_win_create( void *base, MPI_Aint size, int disp_unit, MPI_Info info, MPI_Comm comm){
  31
  32   MPI_Win win;
  33
  34   int comm_size = smpi_comm_size(comm);
  35   int rank=smpi_comm_rank(comm);
  36   XBT_DEBUG("Creating window");
  37
  38   win = xbt_new(s_smpi_mpi_win_t, 1);
  39   win->base = base;
  40   win->size = size;
  41   win->disp_unit = disp_unit;
  42   win->assert = 0;
  43   //win->info = info;
  44   win->comm = comm;
  45   win->name = NULL;
  46   win->requests = xbt_dynar_new(sizeof(MPI_Request), NULL);
  47   win->connected_wins = xbt_malloc0(comm_size*sizeof(MPI_Win));
  48   win->connected_wins[rank] = win;
  49
  50   if(rank==0){
  51     win->bar=xbt_barrier_init(comm_size);
  52   }
  53
  54   mpi_coll_allgather_fun(&(win->connected_wins[rank]),
  55                      sizeof(MPI_Win),
  56                      MPI_BYTE,
  57                      win->connected_wins,
  58                      sizeof(MPI_Win),
  59                      MPI_BYTE,
  60                      comm);
  61
  62   mpi_coll_bcast_fun( &(win->bar),
  63                      sizeof(xbt_bar_t),
  64                      MPI_BYTE,
  65                      0,
  66                      comm);
  67
  68   mpi_coll_barrier_fun(comm);
  69
  70   return win;
  71 }
  72
  73 int smpi_mpi_win_free( MPI_Win* win){
  74
  75   //As per the standard, perform a barrier to ensure every async comm is finished
  76   xbt_barrier_wait((*win)->bar);
  77   xbt_dynar_free(&(*win)->requests);
  78   xbt_free((*win)->connected_wins);
  79   if ((*win)->name != NULL){
  80     xbt_free((*win)->name);
  81   }
  82   xbt_free(*win);
  83   win = MPI_WIN_NULL;
  84   return MPI_SUCCESS;
  85 }
  86
  87 void smpi_mpi_win_get_name(MPI_Win win, char* name, int* length){
  88   if(win->name==NULL){
  89     *length=0;
  90     name=NULL;
  91     return;
  92   }
  93   *length = strlen(win->name);
  94   strcpy(name, win->name);
  95 }
  96
  97 void smpi_mpi_win_set_name(MPI_Win win, char* name){
  98   win->name = strdup(name);;
  99 }
 100
 101
 102 int smpi_mpi_win_fence( int assert,  MPI_Win win){
 103
 104   XBT_DEBUG("Entering fence");
 105
 106   if(assert != MPI_MODE_NOPRECEDE){
 107     xbt_barrier_wait(win->bar);
 108
 109     xbt_dynar_t reqs = win->requests;
 110     int size = xbt_dynar_length(reqs);
 111     unsigned int cpt=0;
 112     MPI_Request req;
 113     // start all requests that have been prepared by another process
 114     xbt_dynar_foreach(reqs, cpt, req){
 115       if (req->flags & PREPARED) smpi_mpi_start(req);
 116     }
 117
 118     MPI_Request* treqs = xbt_dynar_to_array(reqs);
 119     smpi_mpi_waitall(size,treqs,MPI_STATUSES_IGNORE);
 120     xbt_free(treqs);
 121     win->requests=xbt_dynar_new(sizeof(MPI_Request), NULL);
 122
 123   }
 124   win->assert = assert;
 125
 126   xbt_barrier_wait(win->bar);
 127   XBT_DEBUG("Leaving fence ");
 128
 129   return MPI_SUCCESS;
 130 }
 131
 132 int smpi_mpi_put( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
 133               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Win win)
 134 {
 135   //get receiver pointer
 136   MPI_Win recv_win = win->connected_wins[target_rank];
 137
 138   void* recv_addr = (void*) ( ((char*)recv_win->base) + target_disp * recv_win->disp_unit);
 139   smpi_datatype_use(origin_datatype);
 140   smpi_datatype_use(target_datatype);
 141   XBT_DEBUG("Entering MPI_Put to %d", target_rank);
 142
 143   if(target_rank != smpi_comm_rank(win->comm)){
 144     //prepare send_request
 145     MPI_Request sreq = smpi_rma_send_init(origin_addr, origin_count, origin_datatype,
 146         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+1, win->comm, MPI_OP_NULL);
 147
 148     //prepare receiver request
 149     MPI_Request rreq = smpi_rma_recv_init(recv_addr, target_count, target_datatype,
 150         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+1, recv_win->comm, MPI_OP_NULL);
 151
 152     //push request to receiver's win
 153     xbt_dynar_push_as(recv_win->requests, MPI_Request, rreq);
 154
 155     //start send
 156     smpi_mpi_start(sreq);
 157
 158     //push request to sender's win
 159     xbt_dynar_push_as(win->requests, MPI_Request, sreq);
 160   }else{
 161     smpi_datatype_copy(origin_addr, origin_count, origin_datatype,
 162                        recv_addr, target_count, target_datatype);
 163   }
 164
 165   return MPI_SUCCESS;
 166 }
 167
 168 int smpi_mpi_get( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
 169               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Win win)
 170 {
 171   //get sender pointer
 172   MPI_Win send_win = win->connected_wins[target_rank];
 173
 174   void* send_addr = (void*)( ((char*)send_win->base) + target_disp * send_win->disp_unit);
 175   smpi_datatype_use(origin_datatype);
 176   smpi_datatype_use(target_datatype);
 177   XBT_DEBUG("Entering MPI_Get from %d", target_rank);
 178
 179   if(target_rank != smpi_comm_rank(win->comm)){
 180     //prepare send_request
 181     MPI_Request sreq = smpi_rma_send_init(send_addr, target_count, target_datatype,
 182         smpi_group_index(smpi_comm_group(win->comm),target_rank), smpi_process_index(), RMA_TAG+2, send_win->comm, MPI_OP_NULL);
 183
 184     //prepare receiver request
 185     MPI_Request rreq = smpi_rma_recv_init(origin_addr, origin_count, origin_datatype,
 186         smpi_group_index(smpi_comm_group(win->comm),target_rank), smpi_process_index(), RMA_TAG+2, win->comm, MPI_OP_NULL);
 187
 188     //start the send, with another process than us as sender.
 189     smpi_mpi_start(sreq);
 190
 191     //push request to receiver's win
 192     xbt_dynar_push_as(send_win->requests, MPI_Request, sreq);
 193
 194     //start recv
 195     smpi_mpi_start(rreq);
 196
 197     //push request to sender's win
 198     xbt_dynar_push_as(win->requests, MPI_Request, rreq);
 199   }else{
 200     smpi_datatype_copy(send_addr, target_count, target_datatype,
 201                        origin_addr, origin_count, origin_datatype);
 202   }
 203
 204   return MPI_SUCCESS;
 205 }
 206
 207
 208 int smpi_mpi_accumulate( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
 209               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Op op, MPI_Win win)
 210 {
 211   //FIXME: local version
 212   //get receiver pointer
 213   MPI_Win recv_win = win->connected_wins[target_rank];
 214
 215   void* recv_addr = (void*)( ((char*)recv_win->base) + target_disp * recv_win->disp_unit);
 216   XBT_DEBUG("Entering MPI_Accumulate to %d", target_rank);
 217
 218   smpi_datatype_use(origin_datatype);
 219   smpi_datatype_use(target_datatype);
 220
 221
 222     //prepare send_request
 223     MPI_Request sreq = smpi_rma_send_init(origin_addr, origin_count, origin_datatype,
 224         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+3, win->comm, op);
 225
 226     //prepare receiver request
 227     MPI_Request rreq = smpi_rma_recv_init(recv_addr, target_count, target_datatype,
 228         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+3, recv_win->comm, op);
 229     //push request to receiver's win
 230     xbt_dynar_push_as(recv_win->requests, MPI_Request, rreq);
 231     //start send
 232     smpi_mpi_start(sreq);
 233
 234     //push request to sender's win
 235     xbt_dynar_push_as(win->requests, MPI_Request, sreq);
 236
 237
 238
 239   return MPI_SUCCESS;
 240 }
 241