Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
add support of MPI_PROC_NULL, correct behavior of waitall and waitsome for some speci...
[simgrid.git] / src / smpi / smpi_base.c
1 /* Copyright (c) 2007, 2008, 2009, 2010. 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/time.h"
9 #include "mc/mc.h"
10 #include "xbt/replay.h"
11 #include <errno.h>
12 #include "surf/surf.h"
13
14 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi,
15                                 "Logging specific to SMPI (base)");
16
17 static int match_recv(void* a, void* b, smx_action_t ignored) {
18    MPI_Request ref = (MPI_Request)a;
19    MPI_Request req = (MPI_Request)b;
20    XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d",ref->src,req->src, ref->tag, req->tag);
21
22    xbt_assert(ref, "Cannot match recv against null reference");
23    xbt_assert(req, "Cannot match recv against null request");
24    return (ref->src == MPI_ANY_SOURCE || req->src == ref->src)
25           && (ref->tag == MPI_ANY_TAG || req->tag == ref->tag);
26 }
27
28 static int match_send(void* a, void* b,smx_action_t ignored) {
29    MPI_Request ref = (MPI_Request)a;
30    MPI_Request req = (MPI_Request)b;
31    XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d",ref->src,req->src, ref->tag, req->tag);
32    xbt_assert(ref, "Cannot match send against null reference");
33    xbt_assert(req, "Cannot match send against null request");
34    return (req->src == MPI_ANY_SOURCE || req->src == ref->src)
35           && (req->tag == MPI_ANY_TAG || req->tag == ref->tag);
36 }
37
38 static MPI_Request build_request(void *buf, int count,
39                                  MPI_Datatype datatype, int src, int dst,
40                                  int tag, MPI_Comm comm, unsigned flags)
41 {
42   MPI_Request request;
43
44   request = xbt_new(s_smpi_mpi_request_t, 1);
45   request->buf = buf;
46   // FIXME: this will have to be changed to support non-contiguous datatypes
47   request->size = smpi_datatype_size(datatype) * count;
48   request->src = src;
49   request->dst = dst;
50   request->tag = tag;
51   request->comm = comm;
52   request->action = NULL;
53   request->flags = flags;
54 #ifdef HAVE_TRACING
55   request->send = 0;
56   request->recv = 0;
57 #endif
58   return request;
59 }
60
61
62 void smpi_empty_status(MPI_Status * status) {
63   if(status != MPI_STATUS_IGNORE) {
64       status->MPI_SOURCE=MPI_ANY_SOURCE;
65       status->MPI_TAG=MPI_ANY_TAG;
66       status->count=0;
67   }
68 }
69
70 void smpi_action_trace_run(char *path)
71 {
72   char *name;
73   xbt_dynar_t todo;
74   xbt_dict_cursor_t cursor;
75
76   action_fp=NULL;
77   if (path) {
78     action_fp = fopen(path, "r");
79     xbt_assert(action_fp != NULL, "Cannot open %s: %s", path,
80                 strerror(errno));
81   }
82
83   if (!xbt_dict_is_empty(action_queues)) {
84     XBT_WARN
85         ("Not all actions got consumed. If the simulation ended successfully (without deadlock), you may want to add new processes to your deployment file.");
86
87
88     xbt_dict_foreach(action_queues, cursor, name, todo) {
89       XBT_WARN("Still %lu actions for %s", xbt_dynar_length(todo), name);
90     }
91   }
92
93   if (path)
94     fclose(action_fp);
95   xbt_dict_free(&action_queues);
96   action_queues = xbt_dict_new_homogeneous(NULL);
97 }
98
99 static void smpi_mpi_request_free_voidp(void* request)
100 {
101   MPI_Request req = request;
102   smpi_mpi_request_free(&req);
103 }
104
105 /* MPI Low level calls */
106 MPI_Request smpi_mpi_send_init(void *buf, int count, MPI_Datatype datatype,
107                                int dst, int tag, MPI_Comm comm)
108 {
109   MPI_Request request =
110       build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
111                     comm, PERSISTENT | SEND);
112
113   return request;
114 }
115
116 MPI_Request smpi_mpi_recv_init(void *buf, int count, MPI_Datatype datatype,
117                                int src, int tag, MPI_Comm comm)
118 {
119   MPI_Request request =
120       build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag,
121                     comm, PERSISTENT | RECV);
122
123   return request;
124 }
125
126 void smpi_mpi_start(MPI_Request request)
127 {
128   smx_rdv_t mailbox;
129   int detached = 0;
130
131   xbt_assert(!request->action,
132               "Cannot (re)start a non-finished communication");
133   if(request->flags & RECV) {
134     print_request("New recv", request);
135     if (request->size < xbt_cfg_get_int(_surf_cfg_set, "smpi/async_small_thres"))
136     mailbox = smpi_process_mailbox_small();
137     else
138     mailbox = smpi_process_mailbox();
139
140     // FIXME: SIMIX does not yet support non-contiguous datatypes
141     request->action = simcall_comm_irecv(mailbox, request->buf, &request->size, &match_recv, request);
142   } else {
143     print_request("New send", request);
144     if (request->size < xbt_cfg_get_int(_surf_cfg_set, "smpi/async_small_thres")) { // eager mode
145       mailbox = smpi_process_remote_mailbox_small(
146             smpi_group_index(smpi_comm_group(request->comm), request->dst));
147     }else{
148       XBT_DEBUG("Send request %p is not in the permanent receive mailbox (buf: %p)",request,request->buf);
149       mailbox = smpi_process_remote_mailbox(
150                   smpi_group_index(smpi_comm_group(request->comm), request->dst));
151     }
152     if (request->size < 64*1024 ) { //(FIXME: this limit should be configurable)
153       void *oldbuf = request->buf;
154       detached = 1;
155       request->buf = malloc(request->size);
156       if (oldbuf)
157         memcpy(request->buf,oldbuf,request->size);
158       XBT_DEBUG("Send request %p is detached; buf %p copied into %p",request,oldbuf,request->buf);
159     }
160
161       request->action =
162       simcall_comm_isend(mailbox, request->size, -1.0,
163               request->buf, request->size,
164               &match_send,
165               &smpi_mpi_request_free_voidp, // how to free the userdata if a detached send fails
166               request,
167               // detach if msg size < eager/rdv switch limit
168               detached);
169
170   #ifdef HAVE_TRACING
171       /* FIXME: detached sends are not traceable (request->action == NULL) */
172       if (request->action)
173         simcall_set_category(request->action, TRACE_internal_smpi_get_category());
174   #endif
175
176     }
177
178 }
179
180 void smpi_mpi_startall(int count, MPI_Request * requests)
181 {
182     int i;
183
184   for(i = 0; i < count; i++) {
185     smpi_mpi_start(requests[i]);
186   }
187 }
188
189 void smpi_mpi_request_free(MPI_Request * request)
190 {
191   xbt_free(*request);
192   *request = MPI_REQUEST_NULL;
193 }
194
195 MPI_Request smpi_isend_init(void *buf, int count, MPI_Datatype datatype,
196                             int dst, int tag, MPI_Comm comm)
197 {
198   MPI_Request request =
199       build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
200                     comm, NON_PERSISTENT | SEND);
201
202   return request;
203 }
204
205 MPI_Request smpi_mpi_isend(void *buf, int count, MPI_Datatype datatype,
206                            int dst, int tag, MPI_Comm comm)
207 {
208   MPI_Request request =
209       smpi_isend_init(buf, count, datatype, dst, tag, comm);
210
211   smpi_mpi_start(request);
212   return request;
213 }
214
215 MPI_Request smpi_irecv_init(void *buf, int count, MPI_Datatype datatype,
216                             int src, int tag, MPI_Comm comm)
217 {
218   MPI_Request request =
219       build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag,
220                     comm, NON_PERSISTENT | RECV);
221
222   return request;
223 }
224
225 MPI_Request smpi_mpi_irecv(void *buf, int count, MPI_Datatype datatype,
226                            int src, int tag, MPI_Comm comm)
227 {
228   MPI_Request request =
229       smpi_irecv_init(buf, count, datatype, src, tag, comm);
230
231   smpi_mpi_start(request);
232   return request;
233 }
234
235 void smpi_mpi_recv(void *buf, int count, MPI_Datatype datatype, int src,
236                    int tag, MPI_Comm comm, MPI_Status * status)
237 {
238   MPI_Request request;
239
240   request = smpi_mpi_irecv(buf, count, datatype, src, tag, comm);
241   smpi_mpi_wait(&request, status);
242 }
243
244
245
246 void smpi_mpi_send(void *buf, int count, MPI_Datatype datatype, int dst,
247                    int tag, MPI_Comm comm)
248 {
249           MPI_Request request;
250           request = smpi_mpi_isend(buf, count, datatype, dst, tag, comm);
251           smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
252 }
253
254 void smpi_mpi_sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
255                        int dst, int sendtag, void *recvbuf, int recvcount,
256                        MPI_Datatype recvtype, int src, int recvtag,
257                        MPI_Comm comm, MPI_Status * status)
258 {
259   MPI_Request requests[2];
260   MPI_Status stats[2];
261
262   requests[0] =
263       smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
264   requests[1] =
265       smpi_irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm);
266   smpi_mpi_startall(2, requests);
267   smpi_mpi_waitall(2, requests, stats);
268   if(status != MPI_STATUS_IGNORE) {
269     // Copy receive status
270     memcpy(status, &stats[1], sizeof(MPI_Status));
271   }
272 }
273
274 int smpi_mpi_get_count(MPI_Status * status, MPI_Datatype datatype)
275 {
276   return status->count / smpi_datatype_size(datatype);
277 }
278
279 static void finish_wait(MPI_Request * request, MPI_Status * status)
280 {
281   MPI_Request req = *request;
282   // if we have a sender, we should use its data, and not the data from the receive
283   if((req->action)&&
284       (req->src==MPI_ANY_SOURCE || req->tag== MPI_ANY_TAG))
285     req = (MPI_Request)SIMIX_comm_get_src_data((*request)->action);
286
287   if(status != MPI_STATUS_IGNORE) {
288     status->MPI_SOURCE = req->src;
289     status->MPI_TAG = req->tag;
290     status->MPI_ERROR = MPI_SUCCESS;
291     // FIXME: really this should just contain the count of receive-type blocks,
292     // right?
293     status->count = req->size;
294   }
295   req = *request;
296
297   print_request("Finishing", req);
298   if(req->flags & NON_PERSISTENT) {
299     smpi_mpi_request_free(request);
300   } else {
301     req->action = NULL;
302   }
303 }
304
305 int smpi_mpi_test(MPI_Request * request, MPI_Status * status) {
306   int flag;
307   //assume that request is not MPI_REQUEST_NULL (filtered in PMPI_Test or smpi_mpi_testall before)
308   if ((*request)->action == NULL)
309     flag = 1;
310   else
311     flag = simcall_comm_test((*request)->action);
312   if(flag) {
313     finish_wait(request, status);
314   }else{
315     smpi_empty_status(status);
316   }
317   return flag;
318 }
319
320 int smpi_mpi_testany(int count, MPI_Request requests[], int *index,
321                      MPI_Status * status)
322 {
323   xbt_dynar_t comms;
324   int i, flag, size;
325   int* map;
326
327   *index = MPI_UNDEFINED;
328   flag = 0;
329   if(count > 0) {
330     comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
331     map = xbt_new(int, count);
332     size = 0;
333     for(i = 0; i < count; i++) {
334       if((requests[i]!=MPI_REQUEST_NULL) && requests[i]->action) {
335          xbt_dynar_push(comms, &requests[i]->action);
336          map[size] = i;
337          size++;
338       }
339     }
340     if(size > 0) {
341       i = simcall_comm_testany(comms);
342       // not MPI_UNDEFINED, as this is a simix return code
343       if(i != -1) {
344         *index = map[i];
345         finish_wait(&requests[*index], status);
346         flag = 1;
347       }
348     }else{
349         //all requests are null or inactive, return true
350         flag=1;
351         smpi_empty_status(status);
352     }
353     xbt_free(map);
354     xbt_dynar_free(&comms);
355   }
356
357   return flag;
358 }
359
360
361 int smpi_mpi_testall(int count, MPI_Request requests[],
362                      MPI_Status status[])
363 {
364   MPI_Status stat;
365   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
366   int flag=1;
367   int i;
368   for(i=0; i<count; i++){
369     if(requests[i]!= MPI_REQUEST_NULL){
370       if (smpi_mpi_test(&requests[i], pstat)!=1){
371         flag=0;
372       }
373     }else{
374       smpi_empty_status(pstat);
375     }
376     if(status != MPI_STATUSES_IGNORE) {
377       memcpy(&status[i], pstat, sizeof(*pstat));
378     }
379   }
380   return flag;
381 }
382
383 void smpi_mpi_probe(int source, int tag, MPI_Comm comm, MPI_Status* status){
384   int flag=0;
385   //FIXME find another wait to avoid busy waiting ?
386   // the issue here is that we have to wait on a nonexistent comm
387   while(flag==0){
388     smpi_mpi_iprobe(source, tag, comm, &flag, status);
389     XBT_DEBUG("Busy Waiting on probing : %d", flag);
390     if(!flag) {
391       simcall_process_sleep(0.0001);
392     }
393   }
394 }
395
396 void smpi_mpi_iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* status){
397   MPI_Request request =build_request(NULL, 0, MPI_CHAR, source, smpi_comm_rank(comm), tag,
398             comm, NON_PERSISTENT | RECV);
399
400   // behave like a receive, but don't do it
401   smx_rdv_t mailbox;
402
403   print_request("New iprobe", request);
404   // We have to test both mailboxes as we don't know if we will receive one one or another
405     if (xbt_cfg_get_int(_surf_cfg_set, "smpi/async_small_thres")>0){
406         mailbox = smpi_process_mailbox_small();
407         XBT_DEBUG("trying to probe the perm recv mailbox");
408         request->action = simcall_comm_iprobe(mailbox, request->src, request->tag, &match_recv, (void*)request);
409     }
410     if (request->action==NULL){
411         mailbox = smpi_process_mailbox();
412         XBT_DEBUG("trying to probe the other mailbox");
413         request->action = simcall_comm_iprobe(mailbox, request->src, request->tag, &match_recv, (void*)request);
414     }
415
416   if(request->action){
417     MPI_Request req = (MPI_Request)SIMIX_comm_get_src_data(request->action);
418     *flag=true;
419     if(status != MPI_STATUS_IGNORE) {
420       status->MPI_SOURCE = req->src;
421       status->MPI_TAG = req->tag;
422       status->MPI_ERROR = MPI_SUCCESS;
423       status->count = req->size;
424     }
425   }
426   else *flag=false;
427   smpi_mpi_request_free(&request);
428
429   return;
430 }
431
432 void smpi_mpi_wait(MPI_Request * request, MPI_Status * status)
433 {
434   print_request("Waiting", *request);
435   if ((*request)->action != NULL) { // this is not a detached send
436     simcall_comm_wait((*request)->action, -1.0);
437     finish_wait(request, status);
438   }
439   // FIXME for a detached send, finish_wait is not called:
440 }
441
442 int smpi_mpi_waitany(int count, MPI_Request requests[],
443                      MPI_Status * status)
444 {
445   xbt_dynar_t comms;
446   int i, size, index;
447   int *map;
448
449   index = MPI_UNDEFINED;
450   if(count > 0) {
451     // Wait for a request to complete
452     comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
453     map = xbt_new(int, count);
454     size = 0;
455     XBT_DEBUG("Wait for one of");
456     for(i = 0; i < count; i++) {
457       if((requests[i] != MPI_REQUEST_NULL) && (requests[i]->action != NULL)) {
458         print_request("Waiting any ", requests[i]);
459         xbt_dynar_push(comms, &requests[i]->action);
460         map[size] = i;
461         size++;
462       }
463     }
464     if(size > 0) {
465       i = simcall_comm_waitany(comms);
466
467       // not MPI_UNDEFINED, as this is a simix return code
468       if (i != -1) {
469         index = map[i];
470         finish_wait(&requests[index], status);
471       }
472     }
473     xbt_free(map);
474     xbt_dynar_free(&comms);
475   }
476
477   if (index==MPI_UNDEFINED)
478     smpi_empty_status(status);
479
480   return index;
481 }
482
483 void smpi_mpi_waitall(int count, MPI_Request requests[],
484                       MPI_Status status[])
485 {
486   int  index, c;
487   MPI_Status stat;
488   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
489   //tag invalid requests in the set
490   for(c = 0; c < count; c++) {
491     if(requests[c]==MPI_REQUEST_NULL || requests[c]->dst == MPI_PROC_NULL ){
492       if(status != MPI_STATUSES_IGNORE)
493         smpi_empty_status(&status[c]);
494     }else if(requests[c]->src == MPI_PROC_NULL ){
495       if(status != MPI_STATUSES_IGNORE) {
496         smpi_empty_status(&status[c]);
497         status[c].MPI_SOURCE=MPI_PROC_NULL;
498       }
499     }
500   }
501
502   for(c = 0; c < count; c++) {
503       if(MC_IS_ENABLED) {
504         smpi_mpi_wait(&requests[c], pstat);
505         index = c;
506       } else {
507         index = smpi_mpi_waitany(count, requests, pstat);
508         if(index == MPI_UNDEFINED) {
509           break;
510        }
511       if(status != MPI_STATUSES_IGNORE) {
512         memcpy(&status[index], pstat, sizeof(*pstat));
513
514       }
515     }
516   }
517 }
518
519 int smpi_mpi_waitsome(int incount, MPI_Request requests[], int *indices,
520                       MPI_Status status[])
521 {
522   int i, count, index;
523   MPI_Status stat;
524   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
525
526   count = 0;
527   for(i = 0; i < incount; i++)
528   {
529     index=smpi_mpi_waitany(incount, requests, pstat);
530     if(index!=MPI_UNDEFINED){
531       indices[count] = index;
532       count++;
533       if(status != MPI_STATUSES_IGNORE) {
534         memcpy(&status[index], pstat, sizeof(*pstat));
535       }
536     }else{
537       return MPI_UNDEFINED;
538     }
539   }
540   return count;
541 }
542
543 int smpi_mpi_testsome(int incount, MPI_Request requests[], int *indices,
544                       MPI_Status status[])
545 {
546   int i, count, count_dead;
547   MPI_Status stat;
548   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
549
550   count = 0;
551   for(i = 0; i < incount; i++) {
552     if((requests[i] != MPI_REQUEST_NULL)) {
553       if(smpi_mpi_test(&requests[i], pstat)) {
554          indices[count] = i;
555          count++;
556          if(status != MPI_STATUSES_IGNORE) {
557             memcpy(&status[i], pstat, sizeof(*pstat));
558          }
559       }
560     }else{
561       count_dead++;
562     }
563   }
564   if(count_dead==incount)return MPI_UNDEFINED;
565   else return count;
566 }
567
568 void smpi_mpi_bcast(void *buf, int count, MPI_Datatype datatype, int root,
569                     MPI_Comm comm)
570 {
571   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
572   nary_tree_bcast(buf, count, datatype, root, comm, 4);
573 }
574
575 void smpi_mpi_barrier(MPI_Comm comm)
576 {
577   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
578   nary_tree_barrier(comm, 4);
579 }
580
581 void smpi_mpi_gather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
582                      void *recvbuf, int recvcount, MPI_Datatype recvtype,
583                      int root, MPI_Comm comm)
584 {
585   int system_tag = 666;
586   int rank, size, src, index;
587   MPI_Aint lb = 0, recvext = 0;
588   MPI_Request *requests;
589
590   rank = smpi_comm_rank(comm);
591   size = smpi_comm_size(comm);
592   if(rank != root) {
593     // Send buffer to root
594     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
595   } else {
596     // FIXME: check for errors
597     smpi_datatype_extent(recvtype, &lb, &recvext);
598     // Local copy from root
599     smpi_datatype_copy(sendbuf, sendcount, sendtype, 
600         (char *)recvbuf + root * recvcount * recvext, recvcount, recvtype);
601     // Receive buffers from senders
602     requests = xbt_new(MPI_Request, size - 1);
603     index = 0;
604     for(src = 0; src < size; src++) {
605       if(src != root) {
606         requests[index] = smpi_irecv_init((char *)recvbuf + src * recvcount * recvext, 
607                                           recvcount, recvtype, 
608                                           src, system_tag, comm);
609         index++;
610       }
611     }
612     // Wait for completion of irecv's.
613     smpi_mpi_startall(size - 1, requests);
614     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
615     xbt_free(requests);
616   }
617 }
618
619 void smpi_mpi_gatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
620                       void *recvbuf, int *recvcounts, int *displs,
621                       MPI_Datatype recvtype, int root, MPI_Comm comm)
622 {
623   int system_tag = 666;
624   int rank, size, src, index;
625   MPI_Aint lb = 0, recvext = 0;
626   MPI_Request *requests;
627
628   rank = smpi_comm_rank(comm);
629   size = smpi_comm_size(comm);
630   if(rank != root) {
631     // Send buffer to root
632     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
633   } else {
634     // FIXME: check for errors
635     smpi_datatype_extent(recvtype, &lb, &recvext);
636     // Local copy from root
637     smpi_datatype_copy(sendbuf, sendcount, sendtype, 
638                        (char *)recvbuf + displs[root] * recvext, 
639                        recvcounts[root], recvtype);
640     // Receive buffers from senders
641     requests = xbt_new(MPI_Request, size - 1);
642     index = 0;
643     for(src = 0; src < size; src++) {
644       if(src != root) {
645         requests[index] =
646             smpi_irecv_init((char *)recvbuf + displs[src] * recvext, 
647                             recvcounts[src], recvtype, src, system_tag, comm);
648         index++;
649       }
650     }
651     // Wait for completion of irecv's.
652     smpi_mpi_startall(size - 1, requests);
653     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
654     xbt_free(requests);
655   }
656 }
657
658 void smpi_mpi_allgather(void *sendbuf, int sendcount,
659                         MPI_Datatype sendtype, void *recvbuf,
660                         int recvcount, MPI_Datatype recvtype,
661                         MPI_Comm comm)
662 {
663   int system_tag = 666;
664   int rank, size, other, index;
665   MPI_Aint lb = 0, recvext = 0;
666   MPI_Request *requests;
667
668   rank = smpi_comm_rank(comm);
669   size = smpi_comm_size(comm);
670   // FIXME: check for errors
671   smpi_datatype_extent(recvtype, &lb, &recvext);
672   // Local copy from self
673   smpi_datatype_copy(sendbuf, sendcount, sendtype, 
674                      (char *)recvbuf + rank * recvcount * recvext, recvcount, 
675                      recvtype);
676   // Send/Recv buffers to/from others;
677   requests = xbt_new(MPI_Request, 2 * (size - 1));
678   index = 0;
679   for(other = 0; other < size; other++) {
680     if(other != rank) {
681       requests[index] =
682           smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag,
683                           comm);
684       index++;
685       requests[index] = smpi_irecv_init((char *)recvbuf + other * recvcount * recvext, 
686                                         recvcount, recvtype, other, 
687                                         system_tag, comm);
688       index++;
689     }
690   }
691   // Wait for completion of all comms.
692   smpi_mpi_startall(2 * (size - 1), requests);
693   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
694   xbt_free(requests);
695 }
696
697 void smpi_mpi_allgatherv(void *sendbuf, int sendcount,
698                          MPI_Datatype sendtype, void *recvbuf,
699                          int *recvcounts, int *displs,
700                          MPI_Datatype recvtype, MPI_Comm comm)
701 {
702   int system_tag = 666;
703   int rank, size, other, index;
704   MPI_Aint lb = 0, recvext = 0;
705   MPI_Request *requests;
706
707   rank = smpi_comm_rank(comm);
708   size = smpi_comm_size(comm);
709   // FIXME: check for errors
710   smpi_datatype_extent(recvtype, &lb, &recvext);
711   // Local copy from self
712   smpi_datatype_copy(sendbuf, sendcount, sendtype, 
713                      (char *)recvbuf + displs[rank] * recvext, 
714                      recvcounts[rank], recvtype);
715   // Send buffers to others;
716   requests = xbt_new(MPI_Request, 2 * (size - 1));
717   index = 0;
718   for(other = 0; other < size; other++) {
719     if(other != rank) {
720       requests[index] =
721           smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag,
722                           comm);
723       index++;
724       requests[index] =
725           smpi_irecv_init((char *)recvbuf + displs[other] * recvext, recvcounts[other],
726                           recvtype, other, system_tag, comm);
727       index++;
728     }
729   }
730   // Wait for completion of all comms.
731   smpi_mpi_startall(2 * (size - 1), requests);
732   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
733   xbt_free(requests);
734 }
735
736 void smpi_mpi_scatter(void *sendbuf, int sendcount, MPI_Datatype sendtype,
737                       void *recvbuf, int recvcount, MPI_Datatype recvtype,
738                       int root, MPI_Comm comm)
739 {
740   int system_tag = 666;
741   int rank, size, dst, index;
742   MPI_Aint lb = 0, sendext = 0;
743   MPI_Request *requests;
744
745   rank = smpi_comm_rank(comm);
746   size = smpi_comm_size(comm);
747   if(rank != root) {
748     // Recv buffer from root
749     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm,
750                   MPI_STATUS_IGNORE);
751   } else {
752     // FIXME: check for errors
753     smpi_datatype_extent(sendtype, &lb, &sendext);
754     // Local copy from root
755     smpi_datatype_copy((char *)sendbuf + root * sendcount * sendext,
756       sendcount, sendtype, recvbuf, recvcount, recvtype);
757     // Send buffers to receivers
758     requests = xbt_new(MPI_Request, size - 1);
759     index = 0;
760     for(dst = 0; dst < size; dst++) {
761       if(dst != root) {
762         requests[index] = smpi_isend_init((char *)sendbuf + dst * sendcount * sendext, 
763                                           sendcount, sendtype, dst,
764                                           system_tag, comm);
765         index++;
766       }
767     }
768     // Wait for completion of isend's.
769     smpi_mpi_startall(size - 1, requests);
770     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
771     xbt_free(requests);
772   }
773 }
774
775 void smpi_mpi_scatterv(void *sendbuf, int *sendcounts, int *displs,
776                        MPI_Datatype sendtype, void *recvbuf, int recvcount,
777                        MPI_Datatype recvtype, int root, MPI_Comm comm)
778 {
779   int system_tag = 666;
780   int rank, size, dst, index;
781   MPI_Aint lb = 0, sendext = 0;
782   MPI_Request *requests;
783
784   rank = smpi_comm_rank(comm);
785   size = smpi_comm_size(comm);
786   if(rank != root) {
787     // Recv buffer from root
788     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm,
789                   MPI_STATUS_IGNORE);
790   } else {
791     // FIXME: check for errors
792     smpi_datatype_extent(sendtype, &lb, &sendext);
793     // Local copy from root
794     smpi_datatype_copy((char *)sendbuf + displs[root] * sendext, sendcounts[root], 
795                        sendtype, recvbuf, recvcount, recvtype);
796     // Send buffers to receivers
797     requests = xbt_new(MPI_Request, size - 1);
798     index = 0;
799     for(dst = 0; dst < size; dst++) {
800       if(dst != root) {
801         requests[index] =
802             smpi_isend_init((char *)sendbuf + displs[dst] * sendext, sendcounts[dst], 
803                             sendtype, dst, system_tag, comm);
804         index++;
805       }
806     }
807     // Wait for completion of isend's.
808     smpi_mpi_startall(size - 1, requests);
809     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
810     xbt_free(requests);
811   }
812 }
813
814 void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count,
815                      MPI_Datatype datatype, MPI_Op op, int root,
816                      MPI_Comm comm)
817 {
818   int system_tag = 666;
819   int rank, size, src, index;
820   MPI_Aint lb = 0, dataext = 0;
821   MPI_Request *requests;
822   void **tmpbufs;
823
824   rank = smpi_comm_rank(comm);
825   size = smpi_comm_size(comm);
826   if(rank != root) {
827     // Send buffer to root
828     smpi_mpi_send(sendbuf, count, datatype, root, system_tag, comm);
829   } else {
830     // FIXME: check for errors
831     smpi_datatype_extent(datatype, &lb, &dataext);
832     // Local copy from root
833     if (sendbuf && recvbuf)
834       smpi_datatype_copy(sendbuf, count, datatype, recvbuf, count, datatype);
835     // Receive buffers from senders
836     //TODO: make a MPI_barrier here ?
837     requests = xbt_new(MPI_Request, size - 1);
838     tmpbufs = xbt_new(void *, size - 1);
839     index = 0;
840     for(src = 0; src < size; src++) {
841       if(src != root) {
842         // FIXME: possibly overkill we we have contiguous/noncontiguous data
843         //  mapping...
844         tmpbufs[index] = xbt_malloc(count * dataext);
845         requests[index] =
846             smpi_irecv_init(tmpbufs[index], count, datatype, src,
847                             system_tag, comm);
848         index++;
849       }
850     }
851     // Wait for completion of irecv's.
852     smpi_mpi_startall(size - 1, requests);
853     for(src = 0; src < size - 1; src++) {
854       index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
855       XBT_DEBUG("finished waiting any request with index %d", index);
856       if(index == MPI_UNDEFINED) {
857         break;
858       }
859       if(op) /* op can be MPI_OP_NULL that does nothing */
860         smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
861     }
862     for(index = 0; index < size - 1; index++) {
863       xbt_free(tmpbufs[index]);
864     }
865     xbt_free(tmpbufs);
866     xbt_free(requests);
867   }
868 }
869
870 void smpi_mpi_allreduce(void *sendbuf, void *recvbuf, int count,
871                         MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
872 {
873   smpi_mpi_reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
874   smpi_mpi_bcast(recvbuf, count, datatype, 0, comm);
875 }
876
877 void smpi_mpi_scan(void *sendbuf, void *recvbuf, int count,
878                    MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
879 {
880   int system_tag = 666;
881   int rank, size, other, index;
882   MPI_Aint lb = 0, dataext = 0;
883   MPI_Request *requests;
884   void **tmpbufs;
885
886   rank = smpi_comm_rank(comm);
887   size = smpi_comm_size(comm);
888
889   // FIXME: check for errors
890   smpi_datatype_extent(datatype, &lb, &dataext);
891
892   // Local copy from self
893   smpi_datatype_copy(sendbuf, count, datatype, recvbuf, count, datatype);
894
895   // Send/Recv buffers to/from others;
896   requests = xbt_new(MPI_Request, size - 1);
897   tmpbufs = xbt_new(void *, rank);
898   index = 0;
899   for(other = 0; other < rank; other++) {
900     // FIXME: possibly overkill we we have contiguous/noncontiguous data 
901     // mapping...
902     tmpbufs[index] = xbt_malloc(count * dataext);
903     requests[index] =
904         smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag,
905                         comm);
906     index++;
907   }
908   for(other = rank + 1; other < size; other++) {
909     requests[index] =
910         smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm);
911     index++;
912   }
913   // Wait for completion of all comms.
914   smpi_mpi_startall(size - 1, requests);
915   for(other = 0; other < size - 1; other++) {
916     index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
917     if(index == MPI_UNDEFINED) {
918       break;
919     }
920     if(index < rank) {
921       // #Request is below rank: it's a irecv
922       smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
923     }
924   }
925   for(index = 0; index < rank; index++) {
926     xbt_free(tmpbufs[index]);
927   }
928   xbt_free(tmpbufs);
929   xbt_free(requests);
930 }