Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
653131aba0cff8356cf1eaa671e75e3c59995ac0
[simgrid.git] / 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_get_group(MPI_Win win, MPI_Group* group){
98   if(win->comm != MPI_COMM_NULL)
99     *group = smpi_comm_group(win->comm);
100 }
101
102 void smpi_mpi_win_set_name(MPI_Win win, char* name){
103   win->name = strdup(name);;
104 }
105
106
107 int smpi_mpi_win_fence( int assert,  MPI_Win win){
108
109   XBT_DEBUG("Entering fence");
110
111   if(assert != MPI_MODE_NOPRECEDE){
112     xbt_barrier_wait(win->bar);
113
114     xbt_dynar_t reqs = win->requests;
115     int size = xbt_dynar_length(reqs);
116     unsigned int cpt=0;
117     MPI_Request req;
118     // start all requests that have been prepared by another process
119     xbt_dynar_foreach(reqs, cpt, req){
120       if (req->flags & PREPARED) smpi_mpi_start(req);
121     }
122
123     MPI_Request* treqs = xbt_dynar_to_array(reqs);
124     smpi_mpi_waitall(size,treqs,MPI_STATUSES_IGNORE);
125     xbt_free(treqs);
126     win->requests=xbt_dynar_new(sizeof(MPI_Request), NULL);
127
128   }
129   win->assert = assert;
130   
131   xbt_barrier_wait(win->bar);
132   XBT_DEBUG("Leaving fence ");
133
134   return MPI_SUCCESS;
135 }
136
137 int smpi_mpi_put( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
138               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Win win)
139 {
140   //get receiver pointer
141   MPI_Win recv_win = win->connected_wins[target_rank];
142
143   void* recv_addr = (void*) ( ((char*)recv_win->base) + target_disp * recv_win->disp_unit);
144   smpi_datatype_use(origin_datatype);
145   smpi_datatype_use(target_datatype);
146   XBT_DEBUG("Entering MPI_Put to %d", target_rank);
147
148   if(target_rank != smpi_comm_rank(win->comm)){
149     //prepare send_request
150     MPI_Request sreq = smpi_rma_send_init(origin_addr, origin_count, origin_datatype,
151         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+1, win->comm, MPI_OP_NULL);
152
153     //prepare receiver request
154     MPI_Request rreq = smpi_rma_recv_init(recv_addr, target_count, target_datatype,
155         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+1, recv_win->comm, MPI_OP_NULL);
156
157     //push request to receiver's win
158     xbt_dynar_push_as(recv_win->requests, MPI_Request, rreq);
159
160     //start send
161     smpi_mpi_start(sreq);
162
163     //push request to sender's win
164     xbt_dynar_push_as(win->requests, MPI_Request, sreq);
165   }else{
166     smpi_datatype_copy(origin_addr, origin_count, origin_datatype,
167                        recv_addr, target_count, target_datatype);
168   }
169
170   return MPI_SUCCESS;
171 }
172
173 int smpi_mpi_get( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
174               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Win win)
175 {
176   //get sender pointer
177   MPI_Win send_win = win->connected_wins[target_rank];
178
179   void* send_addr = (void*)( ((char*)send_win->base) + target_disp * send_win->disp_unit);
180   smpi_datatype_use(origin_datatype);
181   smpi_datatype_use(target_datatype);
182   XBT_DEBUG("Entering MPI_Get from %d", target_rank);
183
184   if(target_rank != smpi_comm_rank(win->comm)){
185     //prepare send_request
186     MPI_Request sreq = smpi_rma_send_init(send_addr, target_count, target_datatype,
187         smpi_group_index(smpi_comm_group(win->comm),target_rank), smpi_process_index(), RMA_TAG+2, send_win->comm, MPI_OP_NULL);
188
189     //prepare receiver request
190     MPI_Request rreq = smpi_rma_recv_init(origin_addr, origin_count, origin_datatype,
191         smpi_group_index(smpi_comm_group(win->comm),target_rank), smpi_process_index(), RMA_TAG+2, win->comm, MPI_OP_NULL);
192         
193     //start the send, with another process than us as sender. 
194     smpi_mpi_start(sreq);
195     
196     //push request to receiver's win
197     xbt_dynar_push_as(send_win->requests, MPI_Request, sreq);
198
199     //start recv
200     smpi_mpi_start(rreq);
201
202     //push request to sender's win
203     xbt_dynar_push_as(win->requests, MPI_Request, rreq);
204   }else{
205     smpi_datatype_copy(send_addr, target_count, target_datatype,
206                        origin_addr, origin_count, origin_datatype);
207   }
208
209   return MPI_SUCCESS;
210 }
211
212
213 int smpi_mpi_accumulate( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
214               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Op op, MPI_Win win)
215 {
216   //FIXME: local version 
217   //get receiver pointer
218   MPI_Win recv_win = win->connected_wins[target_rank];
219
220   void* recv_addr = (void*)( ((char*)recv_win->base) + target_disp * recv_win->disp_unit);
221   XBT_DEBUG("Entering MPI_Accumulate to %d", target_rank);
222
223   smpi_datatype_use(origin_datatype);
224   smpi_datatype_use(target_datatype);
225
226
227     //prepare send_request
228     MPI_Request sreq = smpi_rma_send_init(origin_addr, origin_count, origin_datatype,
229         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+3, win->comm, op);
230
231     //prepare receiver request
232     MPI_Request rreq = smpi_rma_recv_init(recv_addr, target_count, target_datatype,
233         smpi_process_index(), smpi_group_index(smpi_comm_group(win->comm),target_rank), RMA_TAG+3, recv_win->comm, op);
234     //push request to receiver's win
235     xbt_dynar_push_as(recv_win->requests, MPI_Request, rreq);
236     //start send
237     smpi_mpi_start(sreq);
238     
239     //push request to sender's win
240     xbt_dynar_push_as(win->requests, MPI_Request, sreq);
241   
242
243
244   return MPI_SUCCESS;
245 }
246