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