Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Merge branch 'master' of git+ssh://scm.gforge.inria.fr//gitroot/simgrid/simgrid
[simgrid.git] / src / smpi / smpi_win.cpp
1 /* Copyright (c) 2007-2015. The SimGrid Team.
2  * All rights reserved.                                                     */
3
4 /* This program is free software; you can redistribute it and/or modify it
5  * under the terms of the license (GNU LGPL) which comes with this package. */
6
7 #include "private.h"
8 #include <vector>
9
10 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_rma, smpi, "Logging specific to SMPI (RMA operations)");
11
12 namespace simgrid{
13 namespace smpi{
14 std::unordered_map<int, smpi_key_elem> Win::keyvals_;
15 int Win::keyval_id_=0;
16
17 Win::Win(void *base, MPI_Aint size, int disp_unit, MPI_Info info, MPI_Comm comm): base_(base), size_(size), disp_unit_(disp_unit), assert_(0), info_(info), comm_(comm){
18   int comm_size = comm->size();
19   rank_      = comm->rank();
20   XBT_DEBUG("Creating window");
21   if(info!=MPI_INFO_NULL)
22     info->ref();
23   name_ = nullptr;
24   opened_ = 0;
25   group_ = MPI_GROUP_NULL;
26   requests_ = new std::vector<MPI_Request>();
27   mut_=xbt_mutex_init();
28   lock_mut_=xbt_mutex_init();
29   connected_wins_ = new MPI_Win[comm_size];
30   connected_wins_[rank_] = this;
31   count_ = 0;
32   if(rank_==0){
33     bar_ = MSG_barrier_init(comm_size);
34   }
35   mode_=0;
36
37   comm->add_rma_win(this);
38
39   Colls::allgather(&(connected_wins_[rank_]), sizeof(MPI_Win), MPI_BYTE, connected_wins_, sizeof(MPI_Win),
40                          MPI_BYTE, comm);
41
42   Colls::bcast(&(bar_), sizeof(msg_bar_t), MPI_BYTE, 0, comm);
43
44   Colls::barrier(comm);
45 }
46
47 Win::~Win(){
48   //As per the standard, perform a barrier to ensure every async comm is finished
49   MSG_barrier_wait(bar_);
50
51   int finished = finish_comms();
52   XBT_DEBUG("Win destructor - Finished %d RMA calls", finished);
53
54   delete requests_;
55   delete[] connected_wins_;
56   if (name_ != nullptr){
57     xbt_free(name_);
58   }
59   if(info_!=MPI_INFO_NULL){
60     MPI_Info_free(&info_);
61   }
62
63   comm_->remove_rma_win(this);
64
65   Colls::barrier(comm_);
66   int rank=comm_->rank();
67   if(rank == 0)
68     MSG_barrier_destroy(bar_);
69   xbt_mutex_destroy(mut_);
70   xbt_mutex_destroy(lock_mut_);
71
72   cleanup_attr<Win>();
73 }
74
75 void Win::get_name(char* name, int* length){
76   if(name_==nullptr){
77     *length=0;
78     name=nullptr;
79     return;
80   }
81   *length = strlen(name_);
82   strncpy(name, name_, *length+1);
83 }
84
85 void Win::get_group(MPI_Group* group){
86   if(comm_ != MPI_COMM_NULL){
87     *group = comm_->group();
88   } else {
89     *group = MPI_GROUP_NULL;
90   }
91 }
92
93 int Win::rank(){
94   return rank_;
95 }
96
97 MPI_Aint Win::size(){
98   return size_;
99 }
100
101 void* Win::base(){
102   return base_;
103 }
104
105 int Win::disp_unit(){
106   return disp_unit_;
107 }
108
109
110 void Win::set_name(char* name){
111   name_ = xbt_strdup(name);
112 }
113
114 int Win::fence(int assert)
115 {
116   XBT_DEBUG("Entering fence");
117   if (opened_ == 0)
118     opened_=1;
119   if (assert != MPI_MODE_NOPRECEDE) {
120     // This is not the first fence => finalize what came before
121     MSG_barrier_wait(bar_);
122     xbt_mutex_acquire(mut_);
123     // This (simulated) mutex ensures that no process pushes to the vector of requests during the waitall.
124     // Without this, the vector could get redimensionned when another process pushes.
125     // This would result in the array used by Request::waitall() to be invalidated.
126     // Another solution would be to copy the data and cleanup the vector *before* Request::waitall
127     std::vector<MPI_Request> *reqs = requests_;
128     int size = static_cast<int>(reqs->size());
129     // start all requests that have been prepared by another process
130     if (size > 0) {
131       for (const auto& req : *reqs) {
132         if (req && (req->flags() & PREPARED))
133           req->start();
134       }
135
136       MPI_Request* treqs = &(*reqs)[0];
137
138       Request::waitall(size, treqs, MPI_STATUSES_IGNORE);
139     }
140     count_=0;
141     xbt_mutex_release(mut_);
142   }
143   assert_ = assert;
144
145   MSG_barrier_wait(bar_);
146   XBT_DEBUG("Leaving fence");
147
148   return MPI_SUCCESS;
149 }
150
151 int Win::put( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
152               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype)
153 {
154   //get receiver pointer
155   MPI_Win recv_win = connected_wins_[target_rank];
156
157   if(opened_==0){//check that post/start has been done
158     // no fence or start .. lock ok ?
159     int locked=0;
160     for(auto it : recv_win->lockers_)
161       if (it == comm_->rank())
162         locked = 1;
163     if(locked != 1)
164       return MPI_ERR_WIN;
165   }
166
167   if(target_count*target_datatype->get_extent()>recv_win->size_)
168     return MPI_ERR_ARG;
169
170   void* recv_addr = static_cast<void*> ( static_cast<char*>(recv_win->base_) + target_disp * recv_win->disp_unit_);
171   XBT_DEBUG("Entering MPI_Put to %d", target_rank);
172
173   if(target_rank != comm_->rank()){
174     //prepare send_request
175     MPI_Request sreq = Request::rma_send_init(origin_addr, origin_count, origin_datatype, smpi_process()->index(),
176         comm_->group()->index(target_rank), SMPI_RMA_TAG+1, comm_, MPI_OP_NULL);
177
178     //prepare receiver request
179     MPI_Request rreq = Request::rma_recv_init(recv_addr, target_count, target_datatype, smpi_process()->index(),
180         comm_->group()->index(target_rank), SMPI_RMA_TAG+1, recv_win->comm_, MPI_OP_NULL);
181
182     //push request to receiver's win
183     xbt_mutex_acquire(recv_win->mut_);
184     recv_win->requests_->push_back(rreq);
185     xbt_mutex_release(recv_win->mut_);
186     //start send
187     sreq->start();
188
189     //push request to sender's win
190     xbt_mutex_acquire(mut_);
191     requests_->push_back(sreq);
192     xbt_mutex_release(mut_);
193   }else{
194     Datatype::copy(origin_addr, origin_count, origin_datatype, recv_addr, target_count, target_datatype);
195   }
196
197   return MPI_SUCCESS;
198 }
199
200 int Win::get( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
201               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype)
202 {
203   //get sender pointer
204   MPI_Win send_win = connected_wins_[target_rank];
205
206   if(opened_==0){//check that post/start has been done
207     // no fence or start .. lock ok ?
208     int locked=0;
209     for(auto it : send_win->lockers_)
210       if (it == comm_->rank())
211         locked = 1;
212     if(locked != 1)
213       return MPI_ERR_WIN;
214   }
215
216   if(target_count*target_datatype->get_extent()>send_win->size_)
217     return MPI_ERR_ARG;
218
219   void* send_addr = static_cast<void*>(static_cast<char*>(send_win->base_) + target_disp * send_win->disp_unit_);
220   XBT_DEBUG("Entering MPI_Get from %d", target_rank);
221
222   if(target_rank != comm_->rank()){
223     //prepare send_request
224     MPI_Request sreq = Request::rma_send_init(send_addr, target_count, target_datatype,
225         comm_->group()->index(target_rank), smpi_process()->index(), SMPI_RMA_TAG+2, send_win->comm_,
226         MPI_OP_NULL);
227
228     //prepare receiver request
229     MPI_Request rreq = Request::rma_recv_init(origin_addr, origin_count, origin_datatype,
230         comm_->group()->index(target_rank), smpi_process()->index(), SMPI_RMA_TAG+2, comm_,
231         MPI_OP_NULL);
232
233     //start the send, with another process than us as sender. 
234     sreq->start();
235     //push request to receiver's win
236     xbt_mutex_acquire(send_win->mut_);
237     send_win->requests_->push_back(sreq);
238     xbt_mutex_release(send_win->mut_);
239
240     //start recv
241     rreq->start();
242     //push request to sender's win
243     xbt_mutex_acquire(mut_);
244     requests_->push_back(rreq);
245     xbt_mutex_release(mut_);
246   }else{
247     Datatype::copy(send_addr, target_count, target_datatype, origin_addr, origin_count, origin_datatype);
248   }
249
250   return MPI_SUCCESS;
251 }
252
253
254 int Win::accumulate( void *origin_addr, int origin_count, MPI_Datatype origin_datatype, int target_rank,
255               MPI_Aint target_disp, int target_count, MPI_Datatype target_datatype, MPI_Op op)
256 {
257
258   //get receiver pointer
259   MPI_Win recv_win = connected_wins_[target_rank];
260
261   if(opened_==0){//check that post/start has been done
262     // no fence or start .. lock ok ?
263     int locked=0;
264     for(auto it : recv_win->lockers_)
265       if (it == comm_->rank())
266         locked = 1;
267     if(locked != 1)
268       return MPI_ERR_WIN;
269   }
270   //FIXME: local version 
271
272   if(target_count*target_datatype->get_extent()>recv_win->size_)
273     return MPI_ERR_ARG;
274
275   void* recv_addr = static_cast<void*>(static_cast<char*>(recv_win->base_) + target_disp * recv_win->disp_unit_);
276   XBT_DEBUG("Entering MPI_Accumulate to %d", target_rank);
277     //As the tag will be used for ordering of the operations, substract count from it (to avoid collisions with other SMPI tags, SMPI_RMA_TAG is set below all the other ones we use )
278     //prepare send_request
279
280     MPI_Request sreq = Request::rma_send_init(origin_addr, origin_count, origin_datatype,
281         smpi_process()->index(), comm_->group()->index(target_rank), SMPI_RMA_TAG-3-count_, comm_, op);
282
283     //prepare receiver request
284     MPI_Request rreq = Request::rma_recv_init(recv_addr, target_count, target_datatype,
285         smpi_process()->index(), comm_->group()->index(target_rank), SMPI_RMA_TAG-3-count_, recv_win->comm_, op);
286
287     count_++;
288     //push request to receiver's win
289     xbt_mutex_acquire(recv_win->mut_);
290     recv_win->requests_->push_back(rreq);
291     xbt_mutex_release(recv_win->mut_);
292     //start send
293     sreq->start();
294
295     //push request to sender's win
296     xbt_mutex_acquire(mut_);
297     requests_->push_back(sreq);
298     xbt_mutex_release(mut_);
299
300   return MPI_SUCCESS;
301 }
302
303 int Win::start(MPI_Group group, int assert){
304     /* From MPI forum advices
305     The call to MPI_WIN_COMPLETE does not return until the put call has completed at the origin; and the target window
306     will be accessed by the put operation only after the call to MPI_WIN_START has matched a call to MPI_WIN_POST by
307     the target process. This still leaves much choice to implementors. The call to MPI_WIN_START can block until the
308     matching call to MPI_WIN_POST occurs at all target processes. One can also have implementations where the call to
309     MPI_WIN_START is nonblocking, but the call to MPI_PUT blocks until the matching call to MPI_WIN_POST occurred; or
310     implementations where the first two calls are nonblocking, but the call to MPI_WIN_COMPLETE blocks until the call
311     to MPI_WIN_POST occurred; or even implementations where all three calls can complete before any target process
312     called MPI_WIN_POST --- the data put must be buffered, in this last case, so as to allow the put to complete at the
313     origin ahead of its completion at the target. However, once the call to MPI_WIN_POST is issued, the sequence above
314     must complete, without further dependencies.  */
315
316   //naive, blocking implementation.
317     int i             = 0;
318     int j             = 0;
319     int size          = group->size();
320     MPI_Request* reqs = xbt_new0(MPI_Request, size);
321
322     while (j != size) {
323       int src = group->index(j);
324       if (src != smpi_process()->index() && src != MPI_UNDEFINED) {
325         reqs[i] = Request::irecv_init(nullptr, 0, MPI_CHAR, src, SMPI_RMA_TAG + 4, MPI_COMM_WORLD);
326         i++;
327       }
328       j++;
329   }
330   size=i;
331   Request::startall(size, reqs);
332   Request::waitall(size, reqs, MPI_STATUSES_IGNORE);
333   for(i=0;i<size;i++){
334     Request::unref(&reqs[i]);
335   }
336   xbt_free(reqs);
337   opened_++; //we're open for business !
338   group_=group;
339   group->ref();
340   return MPI_SUCCESS;
341 }
342
343 int Win::post(MPI_Group group, int assert){
344   //let's make a synchronous send here
345   int i             = 0;
346   int j             = 0;
347   int size = group->size();
348   MPI_Request* reqs = xbt_new0(MPI_Request, size);
349
350   while(j!=size){
351     int dst=group->index(j);
352     if(dst!=smpi_process()->index() && dst!=MPI_UNDEFINED){
353       reqs[i]=Request::send_init(nullptr, 0, MPI_CHAR, dst, SMPI_RMA_TAG+4, MPI_COMM_WORLD);
354       i++;
355     }
356     j++;
357   }
358   size=i;
359
360   Request::startall(size, reqs);
361   Request::waitall(size, reqs, MPI_STATUSES_IGNORE);
362   for(i=0;i<size;i++){
363     Request::unref(&reqs[i]);
364   }
365   xbt_free(reqs);
366   opened_++; //we're open for business !
367   group_=group;
368   group->ref();
369   return MPI_SUCCESS;
370 }
371
372 int Win::complete(){
373   if(opened_==0)
374     xbt_die("Complete called on already opened MPI_Win");
375
376   XBT_DEBUG("Entering MPI_Win_Complete");
377   int i             = 0;
378   int j             = 0;
379   int size = group_->size();
380   MPI_Request* reqs = xbt_new0(MPI_Request, size);
381
382   while(j!=size){
383     int dst=group_->index(j);
384     if(dst!=smpi_process()->index() && dst!=MPI_UNDEFINED){
385       reqs[i]=Request::send_init(nullptr, 0, MPI_CHAR, dst, SMPI_RMA_TAG+5, MPI_COMM_WORLD);
386       i++;
387     }
388     j++;
389   }
390   size=i;
391   XBT_DEBUG("Win_complete - Sending sync messages to %d processes", size);
392   Request::startall(size, reqs);
393   Request::waitall(size, reqs, MPI_STATUSES_IGNORE);
394
395   for(i=0;i<size;i++){
396     Request::unref(&reqs[i]);
397   }
398   xbt_free(reqs);
399
400   int finished = finish_comms();
401   XBT_DEBUG("Win_complete - Finished %d RMA calls", finished);
402
403   Group::unref(group_);
404   opened_--; //we're closed for business !
405   return MPI_SUCCESS;
406 }
407
408 int Win::wait(){
409   //naive, blocking implementation.
410   XBT_DEBUG("Entering MPI_Win_Wait");
411   int i=0,j=0;
412   int size = group_->size();
413   MPI_Request* reqs = xbt_new0(MPI_Request, size);
414
415   while(j!=size){
416     int src=group_->index(j);
417     if(src!=smpi_process()->index() && src!=MPI_UNDEFINED){
418       reqs[i]=Request::irecv_init(nullptr, 0, MPI_CHAR, src,SMPI_RMA_TAG+5, MPI_COMM_WORLD);
419       i++;
420     }
421     j++;
422   }
423   size=i;
424   XBT_DEBUG("Win_wait - Receiving sync messages from %d processes", size);
425   Request::startall(size, reqs);
426   Request::waitall(size, reqs, MPI_STATUSES_IGNORE);
427   for(i=0;i<size;i++){
428     Request::unref(&reqs[i]);
429   }
430   xbt_free(reqs);
431   int finished = finish_comms();
432   XBT_DEBUG("Win_wait - Finished %d RMA calls", finished);
433
434   Group::unref(group_);
435   opened_--; //we're opened for business !
436   return MPI_SUCCESS;
437 }
438
439 int Win::lock(int lock_type, int rank, int assert){
440   if(opened_!=0)
441     return MPI_ERR_WIN;
442
443   MPI_Win target_win = connected_wins_[rank];
444
445   if ((lock_type == MPI_LOCK_EXCLUSIVE && target_win->mode_ != MPI_LOCK_SHARED)|| target_win->mode_ == MPI_LOCK_EXCLUSIVE){
446     xbt_mutex_acquire(target_win->lock_mut_);
447     target_win->mode_+= lock_type;//add the lock_type to differentiate case when we are switching from EXCLUSIVE to SHARED (no release needed in the unlock)
448     if(lock_type == MPI_LOCK_SHARED){//the window used to be exclusive, it's now shared.
449       xbt_mutex_release(target_win->lock_mut_);
450    }
451   } else if(!(target_win->mode_==MPI_LOCK_SHARED && lock_type == MPI_LOCK_EXCLUSIVE))
452         target_win->mode_+= lock_type; // don't set to exclusive if it's already shared
453
454   target_win->lockers_.push_back(comm_->rank());
455
456   int finished = finish_comms();
457   XBT_DEBUG("Win_lock %d - Finished %d RMA calls", rank, finished);
458
459   return MPI_SUCCESS;
460 }
461
462 int Win::unlock(int rank){
463   if(opened_!=0)
464     return MPI_ERR_WIN;
465
466   MPI_Win target_win = connected_wins_[rank];
467   int target_mode = target_win->mode_;
468   target_win->mode_= 0;
469   target_win->lockers_.remove(comm_->rank());
470   if (target_mode==MPI_LOCK_EXCLUSIVE){
471     xbt_mutex_release(target_win->lock_mut_);
472   }
473
474   int finished = finish_comms();
475   XBT_DEBUG("Win_unlock %d - Finished %d RMA calls", rank, finished);
476
477   return MPI_SUCCESS;
478 }
479
480 Win* Win::f2c(int id){
481   return static_cast<Win*>(F2C::f2c(id));
482 }
483
484
485 int Win::finish_comms(){
486   xbt_mutex_acquire(mut_);
487   //Finish own requests
488   std::vector<MPI_Request> *reqqs = requests_;
489   int size = static_cast<int>(reqqs->size());
490   if (size > 0) {
491     // start all requests that have been prepared by another process
492     for (const auto& req : *reqqs) {
493       if (req && (req->flags() & PREPARED))
494         req->start();
495     }
496
497     MPI_Request* treqs = &(*reqqs)[0];
498     Request::waitall(size, treqs, MPI_STATUSES_IGNORE);
499     reqqs->clear();
500   }
501   xbt_mutex_release(mut_);
502   return size;
503 }
504
505
506 }
507 }