Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
add some comments to avoid that someone breaks the MC with a 'trivial cleanup'
[simgrid.git] / src / smpi / smpi_base.c
1 /* Copyright (c) 2007-2013. 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 "xbt/virtu.h"
9 #include "mc/mc.h"
10 #include "xbt/replay.h"
11 #include <errno.h>
12 #include "simix/smx_private.h"
13 #include "surf/surf.h"
14 #include "simgrid/sg_config.h"
15 #include "colls/colls.h"
16
17 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)");
18
19
20 static int match_recv(void* a, void* b, smx_action_t ignored) {
21    MPI_Request ref = (MPI_Request)a;
22    MPI_Request req = (MPI_Request)b;
23    XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d",ref->src,req->src, ref->tag, req->tag);
24
25   xbt_assert(ref, "Cannot match recv against null reference");
26   xbt_assert(req, "Cannot match recv against null request");
27   if((ref->src == MPI_ANY_SOURCE || req->src == ref->src)
28     && ((ref->tag == MPI_ANY_TAG && req->tag >=0) || req->tag == ref->tag)){
29     //we match, we can transfer some values
30     // FIXME : move this to the copy function ?
31     if(ref->src == MPI_ANY_SOURCE)ref->real_src = req->src;
32     if(ref->tag == MPI_ANY_TAG)ref->real_tag = req->tag;
33     if(ref->real_size < req->real_size) ref->truncated = 1;
34     if(req->detached==1){
35         ref->detached_sender=req; //tie the sender to the receiver, as it is detached and has to be freed in the receiver
36     }
37     XBT_DEBUG("match succeeded");
38     return 1;
39   }else return 0;
40 }
41
42 static int match_send(void* a, void* b,smx_action_t ignored) {
43    MPI_Request ref = (MPI_Request)a;
44    MPI_Request req = (MPI_Request)b;
45    XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d",ref->src,req->src, ref->tag, req->tag);
46    xbt_assert(ref, "Cannot match send against null reference");
47    xbt_assert(req, "Cannot match send against null request");
48
49    if((req->src == MPI_ANY_SOURCE || req->src == ref->src)
50              && ((req->tag == MPI_ANY_TAG && ref->tag >=0)|| req->tag == ref->tag))
51    {
52      if(req->src == MPI_ANY_SOURCE)req->real_src = ref->src;
53      if(req->tag == MPI_ANY_TAG)req->real_tag = ref->tag;
54      if(req->real_size < ref->real_size) req->truncated = 1;
55      if(ref->detached==1){
56          req->detached_sender=ref; //tie the sender to the receiver, as it is detached and has to be freed in the receiver
57      }
58     XBT_DEBUG("match succeeded");
59      return 1;
60    } else return 0;
61 }
62
63
64 typedef struct s_smpi_factor *smpi_factor_t;
65 typedef struct s_smpi_factor {
66   long factor;
67   int nb_values;
68   double values[4];//arbitrary set to 4
69 } s_smpi_factor_t;
70 xbt_dynar_t smpi_os_values = NULL;
71 xbt_dynar_t smpi_or_values = NULL;
72 xbt_dynar_t smpi_ois_values = NULL;
73
74 // Methods used to parse and store the values for timing injections in smpi
75 // These are taken from surf/network.c and generalized to have more factors
76 // These methods should be merged with those in surf/network.c (moved somewhere in xbt ?)
77
78 static int factor_cmp(const void *pa, const void *pb)
79 {
80   return (((s_smpi_factor_t*)pa)->factor > ((s_smpi_factor_t*)pb)->factor);
81 }
82
83
84 static xbt_dynar_t parse_factor(const char *smpi_coef_string)
85 {
86   char *value = NULL;
87   unsigned int iter = 0;
88   s_smpi_factor_t fact;
89   int i=0;
90   xbt_dynar_t smpi_factor, radical_elements, radical_elements2 = NULL;
91
92   smpi_factor = xbt_dynar_new(sizeof(s_smpi_factor_t), NULL);
93   radical_elements = xbt_str_split(smpi_coef_string, ";");
94   xbt_dynar_foreach(radical_elements, iter, value) {
95     fact.nb_values=0;
96     radical_elements2 = xbt_str_split(value, ":");
97     if (xbt_dynar_length(radical_elements2) <2 || xbt_dynar_length(radical_elements2) > 5)
98       xbt_die("Malformed radical for smpi factor!");
99     for(i =0; i<xbt_dynar_length(radical_elements2);i++ ){
100         if (i==0){
101            fact.factor = atol(xbt_dynar_get_as(radical_elements2, i, char *));
102         }else{
103            fact.values[fact.nb_values] = atof(xbt_dynar_get_as(radical_elements2, i, char *));
104            fact.nb_values++;
105         }
106     }
107
108     xbt_dynar_push_as(smpi_factor, s_smpi_factor_t, fact);
109     XBT_DEBUG("smpi_factor:\t%ld : %d values, first: %f", fact.factor, fact.nb_values ,fact.values[0]);
110     xbt_dynar_free(&radical_elements2);
111   }
112   xbt_dynar_free(&radical_elements);
113   iter=0;
114   xbt_dynar_sort(smpi_factor, &factor_cmp);
115   xbt_dynar_foreach(smpi_factor, iter, fact) {
116     XBT_DEBUG("smpi_factor:\t%ld : %d values, first: %f", fact.factor, fact.nb_values ,fact.values[0]);
117   }
118   return smpi_factor;
119 }
120
121 static double smpi_os(double size)
122 {
123   if (!smpi_os_values) {
124     smpi_os_values = parse_factor(sg_cfg_get_string("smpi/os"));
125     smpi_register_static(smpi_os_values, xbt_dynar_free_voidp);
126   }
127   unsigned int iter = 0;
128   s_smpi_factor_t fact;
129   double current=0.0;
130   xbt_dynar_foreach(smpi_os_values, iter, fact) {
131     if (size <= fact.factor) {
132         XBT_DEBUG("os : %lf <= %ld return %f", size, fact.factor, current);
133       return current;
134     }else{
135       current=fact.values[0]+fact.values[1]*size;
136     }
137   }
138   XBT_DEBUG("os : %lf > %ld return %f", size, fact.factor, current);
139
140   return current;
141 }
142
143 static double smpi_ois(double size)
144 {
145   if (!smpi_ois_values) {
146     smpi_ois_values = parse_factor(sg_cfg_get_string("smpi/ois"));
147     smpi_register_static(smpi_ois_values, xbt_dynar_free_voidp);
148   }
149   unsigned int iter = 0;
150   s_smpi_factor_t fact;
151   double current=0.0;
152   xbt_dynar_foreach(smpi_ois_values, iter, fact) {
153     if (size <= fact.factor) {
154         XBT_DEBUG("ois : %lf <= %ld return %f", size, fact.factor, current);
155       return current;
156     }else{
157       current=fact.values[0]+fact.values[1]*size;
158     }
159   }
160   XBT_DEBUG("ois : %lf > %ld return %f", size, fact.factor, current);
161
162   return current;
163 }
164
165 static double smpi_or(double size)
166 {
167   if (!smpi_or_values) {
168     smpi_or_values = parse_factor(sg_cfg_get_string("smpi/or"));
169     smpi_register_static(smpi_or_values, xbt_dynar_free_voidp);
170   }
171   unsigned int iter = 0;
172   s_smpi_factor_t fact;
173   double current=0.0;
174   xbt_dynar_foreach(smpi_or_values, iter, fact) {
175     if (size <= fact.factor) {
176         XBT_DEBUG("or : %lf <= %ld return %f", size, fact.factor, current);
177       return current;
178     }else
179       current=fact.values[0]+fact.values[1]*size;
180   }
181   XBT_DEBUG("or : %lf > %ld return %f", size, fact.factor, current);
182
183   return current;
184 }
185
186 static MPI_Request build_request(void *buf, int count,
187                                  MPI_Datatype datatype, int src, int dst,
188                                  int tag, MPI_Comm comm, unsigned flags)
189 {
190   MPI_Request request = NULL;
191
192   void *old_buf = NULL;
193
194   request = xbt_new(s_smpi_mpi_request_t, 1);
195
196   s_smpi_subtype_t *subtype = datatype->substruct;
197
198   if(datatype->has_subtype == 1){
199     // This part handles the problem of non-contiguous memory
200     old_buf = buf;
201     buf = count==0 ? NULL : xbt_malloc(count*smpi_datatype_size(datatype));
202     if (flags & SEND) {
203       subtype->serialize(old_buf, buf, count, datatype->substruct);
204     }
205   }
206
207   request->buf = buf;
208   // This part handles the problem of non-contiguous memory (for the
209   // unserialisation at the reception)
210   request->old_buf = old_buf;
211   request->old_type = datatype;
212
213   request->size = smpi_datatype_size(datatype) * count;
214   request->src = src;
215   request->dst = dst;
216   request->tag = tag;
217   request->comm = comm;
218   request->action = NULL;
219   request->flags = flags;
220   request->detached = 0;
221   request->detached_sender = NULL;
222   request->real_src = 0;
223
224   request->truncated = 0;
225   request->real_size = 0;
226   request->real_tag = 0;
227
228   request->refcount=1;
229 #ifdef HAVE_TRACING
230   request->send = 0;
231   request->recv = 0;
232 #endif
233   if (flags & SEND) smpi_datatype_unuse(datatype);
234
235   return request;
236 }
237
238
239 void smpi_empty_status(MPI_Status * status)
240 {
241   if(status != MPI_STATUS_IGNORE) {
242     status->MPI_SOURCE = MPI_ANY_SOURCE;
243     status->MPI_TAG = MPI_ANY_TAG;
244     status->MPI_ERROR = MPI_SUCCESS;
245     status->count=0;
246   }
247 }
248
249 void smpi_action_trace_run(char *path)
250 {
251   char *name;
252   xbt_dynar_t todo;
253   xbt_dict_cursor_t cursor;
254
255   action_fp=NULL;
256   if (path) {
257     action_fp = fopen(path, "r");
258     xbt_assert(action_fp != NULL, "Cannot open %s: %s", path,
259                strerror(errno));
260   }
261
262   if (!xbt_dict_is_empty(action_queues)) {
263     XBT_WARN
264       ("Not all actions got consumed. If the simulation ended successfully (without deadlock), you may want to add new processes to your deployment file.");
265
266
267     xbt_dict_foreach(action_queues, cursor, name, todo) {
268       XBT_WARN("Still %lu actions for %s", xbt_dynar_length(todo), name);
269     }
270   }
271
272   if (path)
273     fclose(action_fp);
274   xbt_dict_free(&action_queues);
275   action_queues = xbt_dict_new_homogeneous(NULL);
276 }
277
278 static void smpi_mpi_request_free_voidp(void* request)
279 {
280   MPI_Request req = request;
281   smpi_mpi_request_free(&req);
282 }
283
284 /* MPI Low level calls */
285 MPI_Request smpi_mpi_send_init(void *buf, int count, MPI_Datatype datatype,
286                                int dst, int tag, MPI_Comm comm)
287 {
288   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
289   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, smpi_process_index(), smpi_group_index(smpi_comm_group(comm), dst), tag,
290                           comm, PERSISTENT | SEND | PREPARED);
291   request->refcount++;
292   return request;
293 }
294
295 MPI_Request smpi_mpi_ssend_init(void *buf, int count, MPI_Datatype datatype,
296                                int dst, int tag, MPI_Comm comm)
297 {
298   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
299   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, smpi_process_index(), smpi_group_index(smpi_comm_group(comm), dst), tag,
300                           comm, PERSISTENT | SSEND | SEND | PREPARED);
301   request->refcount++;
302   return request;
303 }
304
305 MPI_Request smpi_mpi_recv_init(void *buf, int count, MPI_Datatype datatype,
306                                int src, int tag, MPI_Comm comm)
307 {
308   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
309   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : smpi_group_index(smpi_comm_group(comm), src), smpi_process_index(), tag,
310                           comm, PERSISTENT | RECV | PREPARED);
311   request->refcount++;
312   return request;
313 }
314
315 void smpi_mpi_start(MPI_Request request)
316 {
317   smx_rdv_t mailbox;
318
319   xbt_assert(!request->action, "Cannot (re)start a non-finished communication");
320   request->flags &= ~PREPARED;
321   if (request->flags & RECV) {
322     print_request("New recv", request);
323     //FIXME: if receive is posted with a large size, but send is smaller, mailboxes may not match !
324     if (request->size < sg_cfg_get_int("smpi/async_small_thres"))
325       mailbox = smpi_process_mailbox_small();
326     else
327       mailbox = smpi_process_mailbox();
328     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
329     request->real_size=request->size;
330     smpi_datatype_use(request->old_type);
331     smpi_comm_use(request->comm);
332     request->action = simcall_comm_irecv(mailbox, request->buf, &request->real_size, &match_recv, request);
333
334     //integrate pseudo-timing for buffering of small messages, do not bother to execute the simcall if 0
335     double sleeptime = request->detached ? smpi_or(request->size) : 0.0;
336     if(sleeptime!=0.0){
337         simcall_process_sleep(sleeptime);
338         XBT_DEBUG("receiving size of %zu : sleep %lf ", request->size, smpi_or(request->size));
339     }
340
341   } else {
342
343
344     int receiver = request->dst;//smpi_group_index(smpi_comm_group(request->comm), request->dst);
345
346     #ifdef HAVE_TRACING
347       int rank = smpi_process_index();
348       if (TRACE_smpi_view_internals()) {
349         TRACE_smpi_send(rank, rank, receiver,request->size);
350       }
351     #endif
352 /*    if(receiver == MPI_UNDEFINED) {*/
353 /*      XBT_WARN("Trying to send a message to a wrong rank");*/
354 /*      return;*/
355 /*    }*/
356     print_request("New send", request);
357     if (request->size < sg_cfg_get_int("smpi/async_small_thres")) { // eager mode
358       mailbox = smpi_process_remote_mailbox_small(receiver);
359     }else{
360       XBT_DEBUG("Send request %p is not in the permanent receive mailbox (buf: %p)",request,request->buf);
361       mailbox = smpi_process_remote_mailbox(receiver);
362     }
363     if ( (! (request->flags & SSEND)) && (request->size < sg_cfg_get_int("smpi/send_is_detached_thres"))) {
364       void *oldbuf = NULL;
365       request->detached = 1;
366       request->refcount++;
367       if(request->old_type->has_subtype == 0){
368         oldbuf = request->buf;
369         if (oldbuf && request->size!=0){
370           request->buf = xbt_malloc(request->size);
371           memcpy(request->buf,oldbuf,request->size);
372         }
373       }
374       XBT_DEBUG("Send request %p is detached; buf %p copied into %p",request,oldbuf,request->buf);
375     }
376
377     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
378     request->real_size=request->size;
379     smpi_datatype_use(request->old_type);
380     smpi_comm_use(request->comm);
381
382     //if we are giving back the control to the user without waiting for completion, we have to inject timings
383     double sleeptime =0.0;
384     if(request->detached || (request->flags & (ISEND|SSEND))){// issend should be treated as isend
385       //isend and send timings may be different
386       sleeptime = (request->flags & ISEND)? smpi_ois(request->size) : smpi_os(request->size);
387     }
388
389     if(sleeptime!=0.0){
390         simcall_process_sleep(sleeptime);
391         XBT_DEBUG("sending size of %zu : sleep %lf ", request->size, smpi_os(request->size));
392     }
393
394     request->action =
395       simcall_comm_isend(mailbox, request->size, -1.0,
396                          request->buf, request->real_size,
397                          &match_send,
398                          &smpi_mpi_request_free_voidp, // how to free the userdata if a detached send fails
399                          request,
400                          // detach if msg size < eager/rdv switch limit
401                          request->detached);
402
403 #ifdef HAVE_TRACING
404     /* FIXME: detached sends are not traceable (request->action == NULL) */
405     if (request->action)
406       simcall_set_category(request->action, TRACE_internal_smpi_get_category());
407
408 #endif
409
410   }
411
412 }
413
414 void smpi_mpi_startall(int count, MPI_Request * requests)
415 {
416   int i;
417
418   for(i = 0; i < count; i++) {
419     smpi_mpi_start(requests[i]);
420   }
421 }
422
423 void smpi_mpi_request_free(MPI_Request * request)
424 {
425   if((*request) != MPI_REQUEST_NULL){
426     (*request)->refcount--;
427     if((*request)->refcount<0) xbt_die("wrong refcount");
428
429     if((*request)->refcount==0){
430         print_request("Destroying", (*request));
431         xbt_free(*request);
432         *request = MPI_REQUEST_NULL;
433     }else{
434         print_request("Decrementing", (*request));
435
436     }
437   }else{
438       xbt_die("freeing an already free request");
439   }
440 }
441
442 MPI_Request smpi_isend_init(void *buf, int count, MPI_Datatype datatype,
443                             int dst, int tag, MPI_Comm comm)
444 {
445   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
446   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf , count, datatype, smpi_process_index(), smpi_group_index(smpi_comm_group(comm), dst), tag,
447                           comm, NON_PERSISTENT | ISEND | SEND | PREPARED);
448   return request;
449 }
450
451 MPI_Request smpi_mpi_isend(void *buf, int count, MPI_Datatype datatype,
452                            int dst, int tag, MPI_Comm comm)
453 {
454   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
455   request =  build_request(buf==MPI_BOTTOM?(void*)0:buf, count, datatype, smpi_process_index(), smpi_group_index(smpi_comm_group(comm), dst), tag,
456                            comm, NON_PERSISTENT | ISEND | SEND);
457   smpi_mpi_start(request);
458   return request;
459 }
460
461 MPI_Request smpi_mpi_issend(void *buf, int count, MPI_Datatype datatype,
462                            int dst, int tag, MPI_Comm comm)
463 {
464   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
465   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, smpi_process_index(), smpi_group_index(smpi_comm_group(comm), dst), tag,
466                           comm, NON_PERSISTENT | ISEND | SSEND | SEND);
467   smpi_mpi_start(request);
468   return request;
469 }
470
471
472
473 MPI_Request smpi_irecv_init(void *buf, int count, MPI_Datatype datatype,
474                             int src, int tag, MPI_Comm comm)
475 {
476   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
477   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : smpi_group_index(smpi_comm_group(comm), src), smpi_process_index(), tag,
478                           comm, NON_PERSISTENT | RECV | PREPARED);
479   return request;
480 }
481
482 MPI_Request smpi_mpi_irecv(void *buf, int count, MPI_Datatype datatype,
483                            int src, int tag, MPI_Comm comm)
484 {
485   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
486   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : smpi_group_index(smpi_comm_group(comm), src), smpi_process_index(), tag,
487                           comm, NON_PERSISTENT | RECV);
488   smpi_mpi_start(request);
489   return request;
490 }
491
492 void smpi_mpi_recv(void *buf, int count, MPI_Datatype datatype, int src,
493                    int tag, MPI_Comm comm, MPI_Status * status)
494 {
495   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
496   request = smpi_mpi_irecv(buf, count, datatype, src, tag, comm);
497   smpi_mpi_wait(&request, status);
498   request = NULL;
499 }
500
501
502
503 void smpi_mpi_send(void *buf, int count, MPI_Datatype datatype, int dst,
504                    int tag, MPI_Comm comm)
505 {
506   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
507   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, smpi_process_index(), smpi_group_index(smpi_comm_group(comm), dst), tag,
508                           comm, NON_PERSISTENT | SEND);
509
510   smpi_mpi_start(request);
511   smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
512   request = NULL;
513 }
514
515 void smpi_mpi_ssend(void *buf, int count, MPI_Datatype datatype,
516                            int dst, int tag, MPI_Comm comm)
517 {
518   MPI_Request request = NULL; /* MC needs the comm to be set to NULL during the call */
519   request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, smpi_process_index(), smpi_group_index(smpi_comm_group(comm), dst), tag,
520                 comm, NON_PERSISTENT | SSEND | SEND);
521
522   smpi_mpi_start(request);
523   smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
524   request = NULL;
525 }
526
527 void smpi_mpi_sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
528                        int dst, int sendtag, void *recvbuf, int recvcount,
529                        MPI_Datatype recvtype, int src, int recvtag,
530                        MPI_Comm comm, MPI_Status * status)
531 {
532   MPI_Request requests[2];
533   MPI_Status stats[2];
534   int myid=smpi_process_index();
535   if ((smpi_group_index(smpi_comm_group(comm), dst) == myid) && (smpi_group_index(smpi_comm_group(comm), src) == myid)) {
536       smpi_datatype_copy(sendbuf, sendcount, sendtype,
537                                      recvbuf, recvcount, recvtype);
538       return;
539   }
540   requests[0] =
541     smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
542   requests[1] =
543     smpi_irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm);
544   smpi_mpi_startall(2, requests);
545   smpi_mpi_waitall(2, requests, stats);
546   if(status != MPI_STATUS_IGNORE) {
547     // Copy receive status
548     *status = stats[1];
549   }
550 }
551
552 int smpi_mpi_get_count(MPI_Status * status, MPI_Datatype datatype)
553 {
554   return status->count / smpi_datatype_size(datatype);
555 }
556
557 static void finish_wait(MPI_Request * request, MPI_Status * status)
558 {
559   MPI_Request req = *request;
560   smpi_empty_status(status);
561
562   if(!(req->detached && req->flags & SEND) && !(req->flags & PREPARED)){
563     if(status != MPI_STATUS_IGNORE) {
564       int src = req->src == MPI_ANY_SOURCE ? req->real_src : req->src;
565       status->MPI_SOURCE = smpi_group_rank(smpi_comm_group(req->comm), src);
566       status->MPI_TAG = req->tag == MPI_ANY_TAG ? req->real_tag : req->tag;
567       status->MPI_ERROR = req->truncated ? MPI_ERR_TRUNCATE : MPI_SUCCESS;
568       // this handles the case were size in receive differs from size in send
569       // FIXME: really this should just contain the count of receive-type blocks,
570       // right?
571       status->count = req->real_size;
572     }
573
574     print_request("Finishing", req);
575     MPI_Datatype datatype = req->old_type;
576
577     if(datatype->has_subtype == 1){
578       // This part handles the problem of non-contignous memory
579       // the unserialization at the reception
580       s_smpi_subtype_t *subtype = datatype->substruct;
581       if(req->flags & RECV) {
582         subtype->unserialize(req->buf, req->old_buf, req->real_size/smpi_datatype_size(datatype) , datatype->substruct);
583       }
584       if(req->detached == 0) free(req->buf);
585     }
586     smpi_comm_unuse(req->comm);
587     smpi_datatype_unuse(datatype);
588
589   }
590
591 #ifdef HAVE_TRACING
592   if (TRACE_smpi_view_internals()) {
593     if(req->flags & RECV){
594       int rank = smpi_process_index();
595       int src_traced = (req->src == MPI_ANY_SOURCE ? req->real_src : req->src);
596       TRACE_smpi_recv(rank, src_traced, rank);
597     }
598   }
599 #endif
600
601   if(req->detached_sender!=NULL){
602     smpi_mpi_request_free(&(req->detached_sender));
603   }
604
605   if(req->flags & NON_PERSISTENT) {
606     smpi_mpi_request_free(request);
607   } else {
608     req->action = NULL;
609   }
610 }
611
612 int smpi_mpi_test(MPI_Request * request, MPI_Status * status) {
613   int flag;
614
615   //assume that request is not MPI_REQUEST_NULL (filtered in PMPI_Test or smpi_mpi_testall before)
616   smpi_empty_status(status);
617   flag = 1;
618   if (!((*request)->flags & PREPARED)) {
619     if ((*request)->action != NULL)
620       flag = simcall_comm_test((*request)->action);
621     if (flag) {
622       finish_wait(request, status);
623       *request = MPI_REQUEST_NULL;
624     }
625   }
626   return flag;
627 }
628
629 int smpi_mpi_testany(int count, MPI_Request requests[], int *index,
630                      MPI_Status * status)
631 {
632   xbt_dynar_t comms;
633   int i, flag, size;
634   int* map;
635
636   *index = MPI_UNDEFINED;
637   flag = 0;
638   comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
639   map = xbt_new(int, count);
640   size = 0;
641   for(i = 0; i < count; i++) {
642     if ((requests[i] != MPI_REQUEST_NULL) && requests[i]->action &&
643         !(requests[i]->flags & PREPARED)) {
644        xbt_dynar_push(comms, &requests[i]->action);
645        map[size] = i;
646        size++;
647     }
648   }
649   if(size > 0) {
650     i = simcall_comm_testany(comms);
651     // not MPI_UNDEFINED, as this is a simix return code
652     if(i != -1) {
653       *index = map[i];
654       finish_wait(&requests[*index], status);
655       requests[*index] = MPI_REQUEST_NULL;
656       flag = 1;
657     }
658   }else{
659       //all requests are null or inactive, return true
660       flag=1;
661       smpi_empty_status(status);
662   }
663   xbt_free(map);
664   xbt_dynar_free(&comms);
665
666   return flag;
667 }
668
669
670 int smpi_mpi_testall(int count, MPI_Request requests[],
671                      MPI_Status status[])
672 {
673   MPI_Status stat;
674   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
675   int flag=1;
676   int i;
677   for(i=0; i<count; i++){
678     if (requests[i] != MPI_REQUEST_NULL && !(requests[i]->flags & PREPARED)) {
679       if (smpi_mpi_test(&requests[i], pstat)!=1){
680         flag=0;
681       }else{
682           requests[i]=MPI_REQUEST_NULL;
683       }
684     }else{
685       smpi_empty_status(pstat);
686     }
687     if(status != MPI_STATUSES_IGNORE) {
688       status[i] = *pstat;
689     }
690   }
691   return flag;
692 }
693
694 void smpi_mpi_probe(int source, int tag, MPI_Comm comm, MPI_Status* status){
695   int flag=0;
696   //FIXME find another wait to avoid busy waiting ?
697   // the issue here is that we have to wait on a nonexistent comm
698   while(flag==0){
699     smpi_mpi_iprobe(source, tag, comm, &flag, status);
700     XBT_DEBUG("Busy Waiting on probing : %d", flag);
701   }
702 }
703
704 void smpi_mpi_iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* status){
705
706   MPI_Request request =build_request(NULL, 0, MPI_CHAR, source == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : smpi_group_index(smpi_comm_group(comm), source), smpi_comm_rank(comm), tag,
707             comm, NON_PERSISTENT | RECV);
708
709   //to avoid deadlock, we have to sleep some time here, or the timer won't advance and we will only do iprobe simcalls
710   double sleeptime= sg_cfg_get_double("smpi/iprobe");
711   //multiplier to the sleeptime, to increase speed of execution, each failed iprobe will increase it
712   static int nsleeps = 1;
713
714   simcall_process_sleep(sleeptime);
715
716   // behave like a receive, but don't do it
717   smx_rdv_t mailbox;
718
719   print_request("New iprobe", request);
720   // We have to test both mailboxes as we don't know if we will receive one one or another
721     if (sg_cfg_get_int("smpi/async_small_thres")>0){
722         mailbox = smpi_process_mailbox_small();
723         XBT_DEBUG("trying to probe the perm recv mailbox");
724         request->action = simcall_comm_iprobe(mailbox, request->src, request->tag, &match_recv, (void*)request);
725     }
726     if (request->action==NULL){
727         mailbox = smpi_process_mailbox();
728         XBT_DEBUG("trying to probe the other mailbox");
729         request->action = simcall_comm_iprobe(mailbox, request->src, request->tag, &match_recv, (void*)request);
730     }
731
732   if(request->action){
733     MPI_Request req = (MPI_Request)SIMIX_comm_get_src_data(request->action);
734     *flag = 1;
735     if(status != MPI_STATUS_IGNORE && !(req->flags & PREPARED)) {
736       status->MPI_SOURCE = smpi_group_rank(smpi_comm_group(comm), req->src);
737       status->MPI_TAG = req->tag;
738       status->MPI_ERROR = MPI_SUCCESS;
739       status->count = req->real_size;
740     }
741     nsleeps=1;//reset the number of sleeps we will do next time
742   }
743   else {
744       *flag = 0;
745       nsleeps++;
746   }
747   smpi_mpi_request_free(&request);
748
749   return;
750 }
751
752 void smpi_mpi_wait(MPI_Request * request, MPI_Status * status)
753 {
754   print_request("Waiting", *request);
755   if ((*request)->flags & PREPARED) {
756     smpi_empty_status(status);
757     return;
758   }
759
760   if ((*request)->action != NULL) { // this is not a detached send
761     simcall_comm_wait((*request)->action, -1.0);
762   }
763
764 #ifdef HAVE_MC
765   if(MC_is_active())
766     (*request)->action->comm.dst_data = NULL; // dangling pointer : dst_data is freed with a wait, need to set it to NULL for system state comparison
767 #endif
768
769   finish_wait(request, status);
770   *request = MPI_REQUEST_NULL;
771   // FIXME for a detached send, finish_wait is not called:
772 }
773
774 int smpi_mpi_waitany(int count, MPI_Request requests[],
775                      MPI_Status * status)
776 {
777   xbt_dynar_t comms;
778   int i, size, index;
779   int *map;
780
781   index = MPI_UNDEFINED;
782   if(count > 0) {
783     // Wait for a request to complete
784     comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
785     map = xbt_new(int, count);
786     size = 0;
787     XBT_DEBUG("Wait for one of %d", count);
788     for(i = 0; i < count; i++) {
789       if (requests[i] != MPI_REQUEST_NULL && !(requests[i]->flags & PREPARED)) {
790         if (requests[i]->action != NULL) {
791           XBT_DEBUG("Waiting any %p ", requests[i]);
792           xbt_dynar_push(comms, &requests[i]->action);
793           map[size] = i;
794           size++;
795         }else{
796          //This is a finished detached request, let's return this one
797          size=0;//so we free the dynar but don't do the waitany call
798          index=i;
799          finish_wait(&requests[i], status);//cleanup if refcount = 0
800          requests[i]=MPI_REQUEST_NULL;//set to null
801          break;
802          }
803       }
804     }
805     if(size > 0) {
806       i = simcall_comm_waitany(comms);
807
808       // not MPI_UNDEFINED, as this is a simix return code
809       if (i != -1) {
810         index = map[i];
811         finish_wait(&requests[index], status);
812         requests[index] = MPI_REQUEST_NULL;
813       }
814     }
815     xbt_free(map);
816     xbt_dynar_free(&comms);
817   }
818
819   if (index==MPI_UNDEFINED)
820     smpi_empty_status(status);
821
822   return index;
823 }
824
825 int smpi_mpi_waitall(int count, MPI_Request requests[],
826                       MPI_Status status[])
827 {
828   int  index, c;
829   MPI_Status stat;
830   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
831   int retvalue = MPI_SUCCESS;
832   //tag invalid requests in the set
833   if (status != MPI_STATUSES_IGNORE) {
834     for (c = 0; c < count; c++) {
835       if (requests[c] == MPI_REQUEST_NULL || requests[c]->dst == MPI_PROC_NULL ||
836           (requests[c]->flags & PREPARED)) {
837         smpi_empty_status(&status[c]);
838       } else if (requests[c]->src == MPI_PROC_NULL) {
839         smpi_empty_status(&status[c]);
840         status[c].MPI_SOURCE = MPI_PROC_NULL;
841       }
842     }
843   }
844   for(c = 0; c < count; c++) {
845     if (MC_is_active()) {
846       smpi_mpi_wait(&requests[c], pstat);
847       index = c;
848     } else {
849       index = smpi_mpi_waitany(count, requests, pstat);
850       if (index == MPI_UNDEFINED)
851         break;
852       requests[index]=MPI_REQUEST_NULL;
853     }
854     if (status != MPI_STATUSES_IGNORE) {
855       status[index] = *pstat;
856       if (status[index].MPI_ERROR == MPI_ERR_TRUNCATE)
857         retvalue = MPI_ERR_IN_STATUS;
858     }
859   }
860
861   return retvalue;
862 }
863
864 int smpi_mpi_waitsome(int incount, MPI_Request requests[], int *indices,
865                       MPI_Status status[])
866 {
867   int i, count, index;
868   MPI_Status stat;
869   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
870
871   count = 0;
872   for(i = 0; i < incount; i++)
873   {
874     index=smpi_mpi_waitany(incount, requests, pstat);
875     if(index!=MPI_UNDEFINED){
876       indices[count] = index;
877       count++;
878       if(status != MPI_STATUSES_IGNORE) {
879         status[index] = *pstat;
880       }
881      requests[index]=MPI_REQUEST_NULL;
882     }else{
883       return MPI_UNDEFINED;
884     }
885   }
886   return count;
887 }
888
889 int smpi_mpi_testsome(int incount, MPI_Request requests[], int *indices,
890                       MPI_Status status[])
891 {
892   int i, count, count_dead;
893   MPI_Status stat;
894   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
895
896   count = 0;
897   count_dead = 0;
898   for(i = 0; i < incount; i++) {
899     if((requests[i] != MPI_REQUEST_NULL)) {
900       if(smpi_mpi_test(&requests[i], pstat)) {
901          indices[i] = 1;
902          count++;
903          if(status != MPI_STATUSES_IGNORE) {
904            status[i] = *pstat;
905          }
906          requests[i]=MPI_REQUEST_NULL;
907
908       }
909     }else{
910       count_dead++;
911     }
912   }
913   if(count_dead==incount)return MPI_UNDEFINED;
914   else return count;
915 }
916
917 void smpi_mpi_bcast(void *buf, int count, MPI_Datatype datatype, int root,
918                     MPI_Comm comm)
919 {
920   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
921   nary_tree_bcast(buf, count, datatype, root, comm, 4);
922 }
923
924 void smpi_mpi_barrier(MPI_Comm comm)
925 {
926   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
927   nary_tree_barrier(comm, 4);
928 }
929
930 void smpi_mpi_gather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
931                      void *recvbuf, int recvcount, MPI_Datatype recvtype,
932                      int root, MPI_Comm comm)
933 {
934   int system_tag = COLL_TAG_GATHER;
935   int rank, size, src, index;
936   MPI_Aint lb = 0, recvext = 0;
937   MPI_Request *requests;
938
939   rank = smpi_comm_rank(comm);
940   size = smpi_comm_size(comm);
941   if(rank != root) {
942     // Send buffer to root
943     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
944   } else {
945     // FIXME: check for errors
946     smpi_datatype_extent(recvtype, &lb, &recvext);
947     // Local copy from root
948     smpi_datatype_copy(sendbuf, sendcount, sendtype,
949                        (char *)recvbuf + root * recvcount * recvext, recvcount, recvtype);
950     // Receive buffers from senders
951     requests = xbt_new(MPI_Request, size - 1);
952     index = 0;
953     for(src = 0; src < size; src++) {
954       if(src != root) {
955         requests[index] = smpi_irecv_init((char *)recvbuf + src * recvcount * recvext,
956                                           recvcount, recvtype,
957                                           src, system_tag, comm);
958         index++;
959       }
960     }
961     // Wait for completion of irecv's.
962     smpi_mpi_startall(size - 1, requests);
963     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
964     xbt_free(requests);
965   }
966 }
967
968
969 void smpi_mpi_reduce_scatter(void *sendbuf, void *recvbuf, int *recvcounts,
970                        MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
971 {
972     int i, size, count;
973     int *displs;
974     int rank = smpi_process_index();
975     void *tmpbuf;
976
977     /* arbitrarily choose root as rank 0 */
978     size = smpi_comm_size(comm);
979     count = 0;
980     displs = xbt_new(int, size);
981     for (i = 0; i < size; i++) {
982       displs[i] = count;
983       count += recvcounts[i];
984     }
985     tmpbuf=(void*)xbt_malloc(count*smpi_datatype_get_extent(datatype));
986     mpi_coll_reduce_fun(sendbuf, tmpbuf, count, datatype, op, 0, comm);
987     smpi_mpi_scatterv(tmpbuf, recvcounts, displs, datatype, recvbuf,
988                       recvcounts[rank], datatype, 0, comm);
989     xbt_free(displs);
990     xbt_free(tmpbuf);
991 }
992
993 void smpi_mpi_gatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
994                       void *recvbuf, int *recvcounts, int *displs,
995                       MPI_Datatype recvtype, int root, MPI_Comm comm)
996 {
997   int system_tag = COLL_TAG_GATHERV;
998   int rank, size, src, index;
999   MPI_Aint lb = 0, recvext = 0;
1000   MPI_Request *requests;
1001
1002   rank = smpi_comm_rank(comm);
1003   size = smpi_comm_size(comm);
1004   if(rank != root) {
1005     // Send buffer to root
1006     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
1007   } else {
1008     // FIXME: check for errors
1009     smpi_datatype_extent(recvtype, &lb, &recvext);
1010     // Local copy from root
1011     smpi_datatype_copy(sendbuf, sendcount, sendtype,
1012                        (char *)recvbuf + displs[root] * recvext,
1013                        recvcounts[root], recvtype);
1014     // Receive buffers from senders
1015     requests = xbt_new(MPI_Request, size - 1);
1016     index = 0;
1017     for(src = 0; src < size; src++) {
1018       if(src != root) {
1019         requests[index] =
1020           smpi_irecv_init((char *)recvbuf + displs[src] * recvext,
1021                           recvcounts[src], recvtype, src, system_tag, comm);
1022         index++;
1023       }
1024     }
1025     // Wait for completion of irecv's.
1026     smpi_mpi_startall(size - 1, requests);
1027     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
1028     xbt_free(requests);
1029   }
1030 }
1031
1032 void smpi_mpi_allgather(void *sendbuf, int sendcount,
1033                         MPI_Datatype sendtype, void *recvbuf,
1034                         int recvcount, MPI_Datatype recvtype,
1035                         MPI_Comm comm)
1036 {
1037   int system_tag = COLL_TAG_ALLGATHER;
1038   int rank, size, other, index;
1039   MPI_Aint lb = 0, recvext = 0;
1040   MPI_Request *requests;
1041
1042   rank = smpi_comm_rank(comm);
1043   size = smpi_comm_size(comm);
1044   // FIXME: check for errors
1045   smpi_datatype_extent(recvtype, &lb, &recvext);
1046   // Local copy from self
1047   smpi_datatype_copy(sendbuf, sendcount, sendtype,
1048                      (char *)recvbuf + rank * recvcount * recvext, recvcount,
1049                      recvtype);
1050   // Send/Recv buffers to/from others;
1051   requests = xbt_new(MPI_Request, 2 * (size - 1));
1052   index = 0;
1053   for(other = 0; other < size; other++) {
1054     if(other != rank) {
1055       requests[index] =
1056         smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag,
1057                         comm);
1058       index++;
1059       requests[index] = smpi_irecv_init((char *)recvbuf + other * recvcount * recvext,
1060                                         recvcount, recvtype, other,
1061                                         system_tag, comm);
1062       index++;
1063     }
1064   }
1065   // Wait for completion of all comms.
1066   smpi_mpi_startall(2 * (size - 1), requests);
1067   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
1068   xbt_free(requests);
1069 }
1070
1071 void smpi_mpi_allgatherv(void *sendbuf, int sendcount,
1072                          MPI_Datatype sendtype, void *recvbuf,
1073                          int *recvcounts, int *displs,
1074                          MPI_Datatype recvtype, MPI_Comm comm)
1075 {
1076   int system_tag = COLL_TAG_ALLGATHERV;
1077   int rank, size, other, index;
1078   MPI_Aint lb = 0, recvext = 0;
1079   MPI_Request *requests;
1080
1081   rank = smpi_comm_rank(comm);
1082   size = smpi_comm_size(comm);
1083   // FIXME: check for errors
1084   smpi_datatype_extent(recvtype, &lb, &recvext);
1085   // Local copy from self
1086   smpi_datatype_copy(sendbuf, sendcount, sendtype,
1087                      (char *)recvbuf + displs[rank] * recvext,
1088                      recvcounts[rank], recvtype);
1089   // Send buffers to others;
1090   requests = xbt_new(MPI_Request, 2 * (size - 1));
1091   index = 0;
1092   for(other = 0; other < size; other++) {
1093     if(other != rank) {
1094       requests[index] =
1095         smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag,
1096                         comm);
1097       index++;
1098       requests[index] =
1099         smpi_irecv_init((char *)recvbuf + displs[other] * recvext, recvcounts[other],
1100                         recvtype, other, system_tag, comm);
1101       index++;
1102     }
1103   }
1104   // Wait for completion of all comms.
1105   smpi_mpi_startall(2 * (size - 1), requests);
1106   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
1107   xbt_free(requests);
1108 }
1109
1110 void smpi_mpi_scatter(void *sendbuf, int sendcount, MPI_Datatype sendtype,
1111                       void *recvbuf, int recvcount, MPI_Datatype recvtype,
1112                       int root, MPI_Comm comm)
1113 {
1114   int system_tag = COLL_TAG_SCATTER;
1115   int rank, size, dst, index;
1116   MPI_Aint lb = 0, sendext = 0;
1117   MPI_Request *requests;
1118
1119   rank = smpi_comm_rank(comm);
1120   size = smpi_comm_size(comm);
1121   if(rank != root) {
1122     // Recv buffer from root
1123     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm,
1124                   MPI_STATUS_IGNORE);
1125   } else {
1126     // FIXME: check for errors
1127     smpi_datatype_extent(sendtype, &lb, &sendext);
1128     // Local copy from root
1129     if(recvbuf!=MPI_IN_PLACE){
1130         smpi_datatype_copy((char *)sendbuf + root * sendcount * sendext,
1131                            sendcount, sendtype, recvbuf, recvcount, recvtype);
1132     }
1133     // Send buffers to receivers
1134     requests = xbt_new(MPI_Request, size - 1);
1135     index = 0;
1136     for(dst = 0; dst < size; dst++) {
1137       if(dst != root) {
1138         requests[index] = smpi_isend_init((char *)sendbuf + dst * sendcount * sendext,
1139                                           sendcount, sendtype, dst,
1140                                           system_tag, comm);
1141         index++;
1142       }
1143     }
1144     // Wait for completion of isend's.
1145     smpi_mpi_startall(size - 1, requests);
1146     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
1147     xbt_free(requests);
1148   }
1149 }
1150
1151 void smpi_mpi_scatterv(void *sendbuf, int *sendcounts, int *displs,
1152                        MPI_Datatype sendtype, void *recvbuf, int recvcount,
1153                        MPI_Datatype recvtype, int root, MPI_Comm comm)
1154 {
1155   int system_tag = COLL_TAG_SCATTERV;
1156   int rank, size, dst, index;
1157   MPI_Aint lb = 0, sendext = 0;
1158   MPI_Request *requests;
1159
1160   rank = smpi_comm_rank(comm);
1161   size = smpi_comm_size(comm);
1162   if(rank != root) {
1163     // Recv buffer from root
1164     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm,
1165                   MPI_STATUS_IGNORE);
1166   } else {
1167     // FIXME: check for errors
1168     smpi_datatype_extent(sendtype, &lb, &sendext);
1169     // Local copy from root
1170     if(recvbuf!=MPI_IN_PLACE){
1171       smpi_datatype_copy((char *)sendbuf + displs[root] * sendext, sendcounts[root],
1172                        sendtype, recvbuf, recvcount, recvtype);
1173     }
1174     // Send buffers to receivers
1175     requests = xbt_new(MPI_Request, size - 1);
1176     index = 0;
1177     for(dst = 0; dst < size; dst++) {
1178       if(dst != root) {
1179         requests[index] =
1180           smpi_isend_init((char *)sendbuf + displs[dst] * sendext, sendcounts[dst],
1181                           sendtype, dst, system_tag, comm);
1182         index++;
1183       }
1184     }
1185     // Wait for completion of isend's.
1186     smpi_mpi_startall(size - 1, requests);
1187     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
1188     xbt_free(requests);
1189   }
1190 }
1191
1192 void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count,
1193                      MPI_Datatype datatype, MPI_Op op, int root,
1194                      MPI_Comm comm)
1195 {
1196   int system_tag = COLL_TAG_REDUCE;
1197   int rank, size, src, index;
1198   MPI_Aint lb = 0, dataext = 0;
1199   MPI_Request *requests;
1200   void **tmpbufs;
1201
1202
1203   char* sendtmpbuf = (char*) sendbuf;
1204   if( sendbuf == MPI_IN_PLACE ) {
1205     sendtmpbuf = (char *)xbt_malloc(count*smpi_datatype_get_extent(datatype));
1206     smpi_datatype_copy(recvbuf, count, datatype,sendtmpbuf, count, datatype);
1207   }
1208
1209   rank = smpi_comm_rank(comm);
1210   size = smpi_comm_size(comm);
1211   //non commutative case, use a working algo from openmpi
1212   if(!smpi_op_is_commute(op)){
1213     smpi_coll_tuned_reduce_ompi_basic_linear(sendtmpbuf, recvbuf, count,
1214                      datatype, op, root, comm);
1215     return;
1216   }
1217   
1218   if(rank != root) {
1219     // Send buffer to root
1220     smpi_mpi_send(sendtmpbuf, count, datatype, root, system_tag, comm);
1221   } else {
1222     // FIXME: check for errors
1223     smpi_datatype_extent(datatype, &lb, &dataext);
1224     // Local copy from root
1225     if (sendtmpbuf && recvbuf)
1226       smpi_datatype_copy(sendtmpbuf, count, datatype, recvbuf, count, datatype);
1227     // Receive buffers from senders
1228     //TODO: make a MPI_barrier here ?
1229     requests = xbt_new(MPI_Request, size - 1);
1230     tmpbufs = xbt_new(void *, size - 1);
1231     index = 0;
1232     for(src = 0; src < size; src++) {
1233       if(src != root) {
1234         // FIXME: possibly overkill we we have contiguous/noncontiguous data
1235         //  mapping...
1236         tmpbufs[index] = xbt_malloc(count * dataext);
1237         requests[index] =
1238           smpi_irecv_init(tmpbufs[index], count, datatype, src,
1239                           system_tag, comm);
1240         index++;
1241       }
1242     }
1243     // Wait for completion of irecv's.
1244     smpi_mpi_startall(size - 1, requests);
1245     for(src = 0; src < size - 1; src++) {
1246       index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
1247       XBT_DEBUG("finished waiting any request with index %d", index);
1248       if(index == MPI_UNDEFINED) {
1249         break;
1250       }
1251       if(op) /* op can be MPI_OP_NULL that does nothing */
1252         smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
1253     }
1254     for(index = 0; index < size - 1; index++) {
1255       xbt_free(tmpbufs[index]);
1256     }
1257     xbt_free(tmpbufs);
1258     xbt_free(requests);
1259
1260     if( sendbuf == MPI_IN_PLACE ) {
1261       xbt_free(sendtmpbuf);
1262     }
1263   }
1264 }
1265
1266 void smpi_mpi_allreduce(void *sendbuf, void *recvbuf, int count,
1267                         MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
1268 {
1269   smpi_mpi_reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
1270   smpi_mpi_bcast(recvbuf, count, datatype, 0, comm);
1271 }
1272
1273 void smpi_mpi_scan(void *sendbuf, void *recvbuf, int count,
1274                    MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
1275 {
1276   int system_tag = -888;
1277   int rank, size, other, index;
1278   MPI_Aint lb = 0, dataext = 0;
1279   MPI_Request *requests;
1280   void **tmpbufs;
1281
1282   rank = smpi_comm_rank(comm);
1283   size = smpi_comm_size(comm);
1284
1285   // FIXME: check for errors
1286   smpi_datatype_extent(datatype, &lb, &dataext);
1287
1288   // Local copy from self
1289   smpi_datatype_copy(sendbuf, count, datatype, recvbuf, count, datatype);
1290
1291   // Send/Recv buffers to/from others;
1292   requests = xbt_new(MPI_Request, size - 1);
1293   tmpbufs = xbt_new(void *, rank);
1294   index = 0;
1295   for(other = 0; other < rank; other++) {
1296     // FIXME: possibly overkill we we have contiguous/noncontiguous data
1297     // mapping...
1298     tmpbufs[index] = xbt_malloc(count * dataext);
1299     requests[index] =
1300       smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag,
1301                       comm);
1302     index++;
1303   }
1304   for(other = rank + 1; other < size; other++) {
1305     requests[index] =
1306       smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm);
1307     index++;
1308   }
1309   // Wait for completion of all comms.
1310   smpi_mpi_startall(size - 1, requests);
1311
1312   if(smpi_op_is_commute(op)){
1313     for(other = 0; other < size - 1; other++) {
1314       index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
1315       if(index == MPI_UNDEFINED) {
1316         break;
1317       }
1318       if(index < rank) {
1319         // #Request is below rank: it's a irecv
1320         smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
1321       }
1322     }
1323   }else{
1324     //non commutative case, wait in order
1325     for(other = 0; other < size - 1; other++) {
1326       smpi_mpi_wait(&(requests[other]), MPI_STATUS_IGNORE);
1327       if(index < rank) {
1328         smpi_op_apply(op, tmpbufs[other], recvbuf, &count, &datatype);
1329       }
1330     }
1331   }
1332   for(index = 0; index < rank; index++) {
1333     xbt_free(tmpbufs[index]);
1334   }
1335   xbt_free(tmpbufs);
1336   xbt_free(requests);
1337 }
1338
1339 void smpi_mpi_exscan(void *sendbuf, void *recvbuf, int count,
1340                    MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
1341 {
1342   int system_tag = -888;
1343   int rank, size, other, index;
1344   MPI_Aint lb = 0, dataext = 0;
1345   MPI_Request *requests;
1346   void **tmpbufs;
1347   int recvbuf_is_empty=1;
1348   rank = smpi_comm_rank(comm);
1349   size = smpi_comm_size(comm);
1350
1351   // FIXME: check for errors
1352   smpi_datatype_extent(datatype, &lb, &dataext);
1353
1354   // Send/Recv buffers to/from others;
1355   requests = xbt_new(MPI_Request, size - 1);
1356   tmpbufs = xbt_new(void *, rank);
1357   index = 0;
1358   for(other = 0; other < rank; other++) {
1359     // FIXME: possibly overkill we we have contiguous/noncontiguous data
1360     // mapping...
1361     tmpbufs[index] = xbt_malloc(count * dataext);
1362     requests[index] =
1363       smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag,
1364                       comm);
1365     index++;
1366   }
1367   for(other = rank + 1; other < size; other++) {
1368     requests[index] =
1369       smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm);
1370     index++;
1371   }
1372   // Wait for completion of all comms.
1373   smpi_mpi_startall(size - 1, requests);
1374   if(smpi_op_is_commute(op)){
1375     for(other = 0; other < size - 1; other++) {
1376       index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
1377       if(index == MPI_UNDEFINED) {
1378         break;
1379       }
1380       if(index < rank) {
1381         if(recvbuf_is_empty){
1382           smpi_datatype_copy(tmpbufs[index], count, datatype, recvbuf, count, datatype);
1383           recvbuf_is_empty=0;
1384         }else
1385         // #Request is below rank: it's a irecv
1386         smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
1387       }
1388     }
1389   }else{
1390     //non commutative case, wait in order
1391     for(other = 0; other < size - 1; other++) {
1392       smpi_mpi_wait(&(requests[other]), MPI_STATUS_IGNORE);
1393       if(index < rank) {
1394           if(recvbuf_is_empty){
1395             smpi_datatype_copy(tmpbufs[other], count, datatype, recvbuf, count, datatype);
1396             recvbuf_is_empty=0;
1397           }else smpi_op_apply(op, tmpbufs[other], recvbuf, &count, &datatype);
1398       }
1399     }
1400   }
1401   for(index = 0; index < rank; index++) {
1402     xbt_free(tmpbufs[index]);
1403   }
1404   xbt_free(tmpbufs);
1405   xbt_free(requests);
1406 }