Logo AND Algorithmique Numérique Distribuée

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