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 /* FIXME:using a global array of MPI_Win simplifies the way to exchange pointers and info,
  15  * but it breaks distributed simulation
  16  */
  17
  18 xbt_bar_t creation_bar = NULL;
  19
  20 typedef struct s_smpi_mpi_win{
  21   void* base;
  22   MPI_Aint size;
  23   int disp_unit;
  24   MPI_Comm comm;
  25   //MPI_Info info
  26   int assert;
  27   xbt_dynar_t requests;
  28   xbt_bar_t bar;
  29   MPI_Win* connected_wins;
  30 } s_smpi_mpi_win_t;
  31
  32
  33 MPI_Win smpi_mpi_win_create( void *base, MPI_Aint size, int disp_unit, MPI_Info info, MPI_Comm comm){
  34
  35   MPI_Win win;
  36
  37   int comm_size = smpi_comm_size(comm);
  38   int rank=smpi_comm_rank(comm);
  39   XBT_DEBUG("Creating window");
  40
  41   win = xbt_new(s_smpi_mpi_win_t, 1);
  42   win->base = base;
  43   win->size = size;
  44   win->disp_unit = disp_unit;
  45   win->assert = 0;
  46   //win->info = info;
  47   win->comm = comm;
  48   win->requests = xbt_dynar_new(sizeof(MPI_Request), NULL);
  49   win->connected_wins = xbt_malloc0(comm_size*sizeof(MPI_Win));
  50   win->connected_wins[rank] = win;
  51
  52   if(rank==0){
  53     win->bar=xbt_barrier_init(comm_size);
  54   }
  55
  56   mpi_coll_allgather_fun(&(win->connected_wins[rank]),
  57                      sizeof(MPI_Win),
  58                      MPI_BYTE,
  59                      win->connected_wins,
  60                      sizeof(MPI_Win),
  61                      MPI_BYTE,
  62                      comm);
  63
  64   mpi_coll_bcast_fun( &(win->bar),
  65                      sizeof(xbt_bar_t),
  66                      MPI_BYTE,
  67                      0,
  68                      comm);
  69
  70   mpi_coll_barrier_fun(comm);
  71
  72   return win;
  73 }
  74
  75 int smpi_mpi_win_free( MPI_Win* win){
  76
  77   //As per the standard, perform a barrier to ensure every async comm is finished
  78   xbt_barrier_wait((*win)->bar);
  79   xbt_dynar_free(&(*win)->requests);
  80   xbt_free((*win)->connected_wins);
  81   xbt_free(*win);
  82   win = MPI_WIN_NULL;
  83   return MPI_SUCCESS;
  84 }
  85
  86
  87 int smpi_mpi_win_fence( int assert,  MPI_Win win){
  88
  89   XBT_DEBUG("Entering fence");
  90
  91   if(assert != MPI_MODE_NOPRECEDE){
  92     xbt_barrier_wait(win->bar);
  93
  94     xbt_dynar_t reqs = win->requests;
  95     int size = xbt_dynar_length(reqs);
  96     unsigned int cpt=0;
  97     MPI_Request req;
  98     // start all requests that have been prepared by another process
  99     xbt_dynar_foreach(reqs, cpt, req){
 100       if (req->flags & PREPARED) smpi_mpi_start(req);
 101     }
 102
 103     MPI_Request* treqs = xbt_dynar_to_array(reqs);
 104     smpi_mpi_waitall(size,treqs,MPI_STATUSES_IGNORE);
 105     xbt_free(treqs);
 106     win->requests=xbt_dynar_new(sizeof(MPI_Request), NULL);
 107
 108   }
 109   win->assert = assert;
 110
 111   xbt_barrier_wait(win->bar);
 112   XBT_DEBUG("Leaving fence ");
 113
 114   return MPI_SUCCESS;
 115 }
 116
 117 int smpi_mpi_put( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
 118               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Win win)
 119 {
 120   //get receiver pointer
 121   MPI_Win recv_win = win->connected_wins[target_rank];
 122
 123   void* recv_addr = recv_win->base + target_disp * smpi_datatype_size(target_datatype)/* recv_win->disp_unit*/;
 124   smpi_datatype_use(origin_datatype);
 125   smpi_datatype_use(target_datatype);
 126   XBT_DEBUG("Entering MPI_Put to %d", target_rank);
 127
 128   if(target_rank != smpi_comm_rank(win->comm)){
 129     //prepare send_request
 130     MPI_Request sreq = smpi_rma_send_init(origin_addr, origin_count, origin_datatype,
 131         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+1, win->comm);
 132
 133     //prepare receiver request
 134     MPI_Request rreq = smpi_rma_recv_init(recv_addr, target_count, target_datatype,
 135         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+1, recv_win->comm);
 136
 137     //push request to receiver's win
 138     xbt_dynar_push_as(recv_win->requests, MPI_Request, rreq);
 139
 140     //start send
 141     smpi_mpi_start(sreq);
 142
 143     //push request to sender's win
 144     xbt_dynar_push_as(win->requests, MPI_Request, sreq);
 145   }
 146   //perform actual copy
 147   /*smpi_datatype_copy(origin_addr, origin_count, origin_datatype,
 148                     recv_addr, target_count, target_datatype);*/
 149
 150   return MPI_SUCCESS;
 151 }
 152
 153 int smpi_mpi_get( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
 154               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Win win)
 155 {
 156   //get sender pointer
 157   MPI_Win send_win = win->connected_wins[target_rank];
 158
 159   void* send_addr = send_win->base + target_disp * smpi_datatype_size(target_datatype)/** send_win->disp_unit*/;
 160   smpi_datatype_use(origin_datatype);
 161   smpi_datatype_use(target_datatype);
 162   XBT_DEBUG("Entering MPI_Get from %d", target_rank);
 163
 164   if(target_rank != smpi_comm_rank(win->comm)){
 165     //prepare send_request
 166     MPI_Request sreq = smpi_rma_send_init(send_addr, target_count, target_datatype,
 167         smpi_group_index(smpi_comm_group(win->comm),target_rank), smpi_process_index(), RMA_TAG+2, send_win->comm);
 168
 169     //prepare receiver request
 170     MPI_Request rreq = smpi_rma_recv_init(origin_addr, origin_count, origin_datatype,
 171         smpi_group_index(smpi_comm_group(win->comm),target_rank), smpi_process_index(), RMA_TAG+2, win->comm);
 172
 173     //push request to receiver's win
 174     xbt_dynar_push_as(send_win->requests, MPI_Request, sreq);
 175
 176     //start recv
 177     smpi_mpi_start(rreq);
 178
 179     //push request to sender's win
 180     xbt_dynar_push_as(win->requests, MPI_Request, rreq);
 181   }
 182   //perform actual copy
 183   /*smpi_datatype_copy(send_addr, target_count, target_datatype,
 184                      origin_addr, origin_count, origin_datatype);*/
 185
 186
 187   return MPI_SUCCESS;
 188 }
 189
 190
 191 int smpi_mpi_accumulate( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
 192               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Op op, MPI_Win win)
 193 {
 194   //get receiver pointer
 195   MPI_Win recv_win = win->connected_wins[target_rank];
 196
 197   void* recv_addr = recv_win->base + target_disp * smpi_datatype_size(target_datatype) /** recv_win->disp_unit*/;
 198   XBT_DEBUG("Entering MPI_Accumulate to %d", target_rank);
 199
 200   smpi_datatype_use(origin_datatype);
 201   smpi_datatype_use(target_datatype);
 202
 203   if(target_rank != smpi_comm_rank(win->comm)){
 204     //prepare send_request
 205     MPI_Request sreq = smpi_rma_send_init(origin_addr, origin_count, origin_datatype,
 206         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+3, win->comm);
 207
 208     //prepare receiver request
 209     MPI_Request rreq = smpi_rma_recv_init(NULL, 0, target_datatype,
 210         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+3, recv_win->comm);
 211     rreq->flags |= ACCUMULATE;
 212     //push request to receiver's win
 213     xbt_dynar_push_as(recv_win->requests, MPI_Request, rreq);
 214     //start send
 215     smpi_mpi_start(sreq);
 216     //push request to sender's win
 217     xbt_dynar_push_as(win->requests, MPI_Request, sreq);
 218    }
 219   //perform actual accumulation
 220   smpi_op_apply(op, origin_addr, recv_addr, &origin_count, &origin_datatype);
 221
 222   return MPI_SUCCESS;
 223 }
 224