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
  10 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi,
  11                                 "Logging specific to SMPI (base)");
  12 XBT_LOG_EXTERNAL_CATEGORY(smpi_base);
  13 XBT_LOG_EXTERNAL_CATEGORY(smpi_bench);
  14 XBT_LOG_EXTERNAL_CATEGORY(smpi_kernel);
  15 XBT_LOG_EXTERNAL_CATEGORY(smpi_mpi);
  16 XBT_LOG_EXTERNAL_CATEGORY(smpi_mpi_dt);
  17 XBT_LOG_EXTERNAL_CATEGORY(smpi_coll);
  18 XBT_LOG_EXTERNAL_CATEGORY(smpi_receiver);
  19 XBT_LOG_EXTERNAL_CATEGORY(smpi_sender);
  20 XBT_LOG_EXTERNAL_CATEGORY(smpi_util);
  21
  22 void smpi_process_init(int* argc, char*** argv) {
  23   int index;
  24   smpi_process_data_t data;
  25   smx_process_t proc;
  26
  27   proc = SIMIX_process_self();
  28   index = atoi((*argv)[1]);
  29   data = smpi_process_remote_data(index);
  30   SIMIX_process_set_data(proc, data);
  31   if (*argc > 2) {
  32     free((*argv)[1]);
  33     memmove(&(*argv)[1], &(*argv)[2], sizeof(char *) * (*argc - 2));
  34     (*argv)[(*argc) - 1] = NULL;
  35   }
  36   (*argc)--;
  37   DEBUG2("<%d> New process in the game: %p", index, proc);
  38 }
  39
  40 void smpi_process_destroy(void) {
  41   int index = smpi_process_index();
  42
  43   DEBUG1("<%d> Process left the game", index);
  44 }
  45
  46 static MPI_Request build_request(void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags) {
  47   MPI_Request request;
  48
  49   request = xbt_new(s_smpi_mpi_request_t, 1);
  50   request->buf = buf;
  51   request->size = smpi_datatype_size(datatype) * count;
  52   request->src = src;
  53   request->dst = dst;
  54   request->tag = tag;
  55   request->comm = comm;
  56   request->rdv = NULL;
  57   request->pair = NULL;
  58   request->complete = 0;
  59   request->match = MPI_REQUEST_NULL;
  60   request->flags = flags;
  61   return request;
  62 }
  63
  64 /* MPI Low level calls */
  65 MPI_Request smpi_mpi_send_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
  66   MPI_Request request = build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag, comm, PERSISTENT | SEND);
  67
  68   return request;
  69 }
  70
  71 MPI_Request smpi_mpi_recv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) {
  72   MPI_Request request = build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag, comm, PERSISTENT | RECV);
  73
  74   return request;
  75 }
  76
  77 void smpi_mpi_start(MPI_Request request) {
  78   xbt_assert0(request->complete == 0, "Cannot start a non-finished communication");
  79   if((request->flags & RECV) == RECV) {
  80     smpi_process_post_recv(request);
  81     print_request("New recv", request);
  82     request->pair = SIMIX_network_irecv(request->rdv, request->buf, &request->size);
  83   } else {
  84     smpi_process_post_send(request->comm, request); // FIXME
  85     print_request("New send", request);
  86     request->pair = SIMIX_network_isend(request->rdv, request->size, -1.0, request->buf, request->size, NULL);
  87   }
  88 }
  89
  90 void smpi_mpi_startall(int count, MPI_Request* requests) {
  91   int i;
  92
  93   for(i = 0; i < count; i++) {
  94     smpi_mpi_start(requests[i]);
  95   }
  96 }
  97
  98 void smpi_mpi_request_free(MPI_Request* request) {
  99   xbt_free(*request);
 100   *request = MPI_REQUEST_NULL;
 101 }
 102
 103 MPI_Request smpi_isend_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
 104   MPI_Request request = build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag, comm, NON_PERSISTENT | SEND);
 105
 106   return request;
 107 }
 108
 109 MPI_Request smpi_mpi_isend(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
 110   MPI_Request request = smpi_isend_init(buf, count, datatype, dst, tag, comm);
 111
 112   smpi_mpi_start(request);
 113   return request;
 114 }
 115
 116 MPI_Request smpi_irecv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) {
 117   MPI_Request request = build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag, comm, NON_PERSISTENT | RECV);
 118
 119   return request;
 120 }
 121
 122 MPI_Request smpi_mpi_irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) {
 123   MPI_Request request = smpi_irecv_init(buf, count, datatype, src, tag, comm);
 124
 125   smpi_mpi_start(request);
 126   return request;
 127 }
 128
 129 void smpi_mpi_recv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status* status) {
 130   MPI_Request request;
 131
 132   request = smpi_mpi_irecv(buf, count, datatype, src, tag, comm);
 133   smpi_mpi_wait(&request, status);
 134 }
 135
 136 void smpi_mpi_send(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
 137   MPI_Request request;
 138
 139   request = smpi_mpi_isend(buf, count, datatype, dst, tag, comm);
 140   smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
 141 }
 142
 143 void smpi_mpi_sendrecv(void* sendbuf, int sendcount, MPI_Datatype sendtype, int dst, int sendtag, void* recvbuf, int recvcount, MPI_Datatype recvtype, int src, int recvtag, MPI_Comm comm, MPI_Status* status) {
 144   MPI_Request requests[2];
 145   MPI_Status stats[2];
 146
 147   requests[0] = smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
 148   requests[1] = smpi_irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm);
 149   smpi_mpi_startall(2, requests);
 150   smpi_mpi_waitall(2, requests, stats);
 151   if(status != MPI_STATUS_IGNORE) {
 152     // Copy receive status
 153     memcpy(status, &stats[1], sizeof(MPI_Status));
 154   }
 155 }
 156
 157 static void finish_wait(MPI_Request* request, MPI_Status* status) {
 158   if(status != MPI_STATUS_IGNORE) {
 159     status->MPI_SOURCE = (*request)->src;
 160     status->MPI_TAG = (*request)->tag;
 161     status->MPI_ERROR = MPI_SUCCESS;
 162   }
 163   print_request("finishing wait", *request);
 164   if((*request)->complete == 1) {
 165     SIMIX_rdv_destroy((*request)->rdv);
 166   } else {
 167     (*request)->match->complete = 1;
 168     (*request)->match->match = MPI_REQUEST_NULL;
 169   }
 170   if(((*request)->flags & NON_PERSISTENT) == NON_PERSISTENT) {
 171     smpi_mpi_request_free(request);
 172   } else {
 173     (*request)->rdv = NULL;
 174     (*request)->pair = NULL;
 175   }
 176 }
 177
 178 int smpi_mpi_test(MPI_Request* request, MPI_Status* status) {
 179   int flag = (*request)->complete;
 180
 181   if(flag) {
 182     SIMIX_communication_destroy((*request)->pair);
 183     finish_wait(request, status);
 184   }
 185   return flag;
 186 }
 187
 188 int smpi_mpi_testany(int count, MPI_Request requests[], int* index, MPI_Status* status) {
 189   int i, flag;
 190
 191   *index = MPI_UNDEFINED;
 192   flag = 0;
 193   for(i = 0; i < count; i++) {
 194     if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) {
 195       SIMIX_communication_destroy(requests[i]->pair);
 196       finish_wait(&requests[i], status);
 197       *index = i;
 198       flag = 1;
 199       break;
 200     }
 201   }
 202   return flag;
 203 }
 204
 205 void smpi_mpi_wait(MPI_Request* request, MPI_Status* status) {
 206   print_request("wait", *request);
 207   // data is null if receiver waits before sender enters the rdv
 208   if((*request)->complete) {
 209     SIMIX_communication_destroy((*request)->pair);
 210   } else {
 211     SIMIX_network_wait((*request)->pair, -1.0);
 212   }
 213   finish_wait(request, status);
 214 }
 215
 216 int smpi_mpi_waitany(int count, MPI_Request requests[], MPI_Status* status) {
 217   xbt_dynar_t comms;
 218   int i, size, index;
 219   int* map;
 220
 221   index = MPI_UNDEFINED;
 222   if(count > 0) {
 223     // First check for already completed requests
 224     for(i = 0; i < count; i++) {
 225       if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) {
 226         index = i;
 227         SIMIX_communication_destroy(requests[index]->pair); // always succeeds (but cleans the simix layer)
 228         break;
 229       }
 230     }
 231     if(index == MPI_UNDEFINED) {
 232       // Otherwise, wait for a request to complete
 233       comms = xbt_dynar_new(sizeof(smx_comm_t), NULL);
 234       map = xbt_new(int, count);
 235       size = 0;
 236       DEBUG0("Wait for one of");
 237       for(i = 0; i < count; i++) {
 238         if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete == 0) {
 239           print_request("   ", requests[i]);
 240           xbt_dynar_push(comms, &requests[i]->pair);
 241           map[size] = i;
 242           size++;
 243         }
 244       }
 245       if(size > 0) {
 246         index = SIMIX_network_waitany(comms);
 247         index = map[index];
 248       }
 249       xbt_free(map);
 250       xbt_dynar_free(&comms);
 251     }
 252     if(index != MPI_UNDEFINED) {
 253       finish_wait(&requests[index], status);
 254     }
 255   }
 256   return index;
 257 }
 258
 259 void smpi_mpi_waitall(int count, MPI_Request requests[],  MPI_Status status[]) {
 260   int index;
 261   MPI_Status stat;
 262
 263   while(count > 0) {
 264     index = smpi_mpi_waitany(count, requests, &stat);
 265     if(index == MPI_UNDEFINED) {
 266       break;
 267     }
 268     if(status != MPI_STATUS_IGNORE) {
 269       memcpy(&status[index], &stat, sizeof(stat));
 270     }
 271     // FIXME: check this -v
 272     // Move the last request to the found position
 273     requests[index] = requests[count - 1];
 274     requests[count - 1] = MPI_REQUEST_NULL;
 275     count--;
 276   }
 277 }
 278
 279 int smpi_mpi_waitsome(int incount, MPI_Request requests[], int* indices, MPI_Status status[]) {
 280   int i, count;
 281
 282   count = 0;
 283   for(i = 0; i < incount; i++) {
 284     if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) {
 285       SIMIX_communication_destroy(requests[i]->pair);
 286       finish_wait(&requests[i], status != MPI_STATUS_IGNORE ? &status[i] : MPI_STATUS_IGNORE);
 287       indices[count] = i;
 288       count++;
 289     }
 290   }
 291   return count;
 292 }
 293
 294 void smpi_mpi_bcast(void* buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm) {
 295   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
 296   nary_tree_bcast(buf, count, datatype, root, comm, 4);
 297 }
 298
 299 void smpi_mpi_barrier(MPI_Comm comm) {
 300   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
 301   nary_tree_barrier(comm, 4);
 302 }
 303
 304 void smpi_mpi_gather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) {
 305   int system_tag = 666;
 306   int rank, size, src, index, sendsize, recvsize;
 307   MPI_Request* requests;
 308
 309   rank = smpi_comm_rank(comm);
 310   size = smpi_comm_size(comm);
 311   if(rank != root) {
 312     // Send buffer to root
 313     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
 314   } else {
 315     sendsize = smpi_datatype_size(sendtype);
 316     recvsize = smpi_datatype_size(recvtype);
 317     // Local copy from root
 318     memcpy(&((char*)recvbuf)[root * recvcount * recvsize], sendbuf, sendcount * sendsize * sizeof(char));
 319     // Receive buffers from senders
 320     requests = xbt_new(MPI_Request, size - 1);
 321     index = 0;
 322     for(src = 0; src < size; src++) {
 323       if(src != root) {
 324         requests[index] = smpi_irecv_init(&((char*)recvbuf)[src * recvcount * recvsize], recvcount, recvtype, src, system_tag, comm);
 325         index++;
 326       }
 327     }
 328     // Wait for completion of irecv's.
 329     smpi_mpi_startall(size - 1, requests);
 330     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 331     xbt_free(requests);
 332   }
 333 }
 334
 335 void smpi_mpi_gatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int* recvcounts, int* displs, MPI_Datatype recvtype, int root, MPI_Comm comm) {
 336   int system_tag = 666;
 337   int rank, size, src, index, sendsize;
 338   MPI_Request* requests;
 339
 340   rank = smpi_comm_rank(comm);
 341   size = smpi_comm_size(comm);
 342   if(rank != root) {
 343     // Send buffer to root
 344     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
 345   } else {
 346     sendsize = smpi_datatype_size(sendtype);
 347     // Local copy from root
 348     memcpy(&((char*)recvbuf)[displs[root]], sendbuf, sendcount * sendsize * sizeof(char));
 349     // Receive buffers from senders
 350     requests = xbt_new(MPI_Request, size - 1);
 351     index = 0;
 352     for(src = 0; src < size; src++) {
 353       if(src != root) {
 354         requests[index] = smpi_irecv_init(&((char*)recvbuf)[displs[src]], recvcounts[src], recvtype, src, system_tag, comm);
 355         index++;
 356       }
 357     }
 358     // Wait for completion of irecv's.
 359     smpi_mpi_startall(size - 1, requests);
 360     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 361     xbt_free(requests);
 362   }
 363 }
 364
 365 void smpi_mpi_allgather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm) {
 366   int system_tag = 666;
 367   int rank, size, other, index, sendsize, recvsize;
 368   MPI_Request* requests;
 369
 370   rank = smpi_comm_rank(comm);
 371   size = smpi_comm_size(comm);
 372   sendsize = smpi_datatype_size(sendtype);
 373   recvsize = smpi_datatype_size(recvtype);
 374   // Local copy from self
 375   memcpy(&((char*)recvbuf)[rank * recvcount * recvsize], sendbuf, sendcount * sendsize * sizeof(char));
 376   // Send/Recv buffers to/from others;
 377   requests = xbt_new(MPI_Request, 2 * (size - 1));
 378   index = 0;
 379   for(other = 0; other < size; other++) {
 380     if(other != rank) {
 381       requests[index] = smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm);
 382       index++;
 383       requests[index] = smpi_irecv_init(&((char*)recvbuf)[other * recvcount * recvsize], recvcount, recvtype, other, system_tag, comm);
 384       index++;
 385     }
 386   }
 387   // Wait for completion of all comms.
 388   smpi_mpi_startall(2 * (size - 1), requests);
 389   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
 390   xbt_free(requests);
 391 }
 392
 393 void smpi_mpi_allgatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int* recvcounts, int* displs, MPI_Datatype recvtype, MPI_Comm comm) {
 394   int system_tag = 666;
 395   int rank, size, other, index, sendsize, recvsize;
 396   MPI_Request* requests;
 397
 398   rank = smpi_comm_rank(comm);
 399   size = smpi_comm_size(comm);
 400   sendsize = smpi_datatype_size(sendtype);
 401   recvsize = smpi_datatype_size(recvtype);
 402   // Local copy from self
 403   memcpy(&((char*)recvbuf)[displs[rank]], sendbuf, sendcount * sendsize * sizeof(char));
 404   // Send buffers to others;
 405   requests = xbt_new(MPI_Request, 2 * (size - 1));
 406   index = 0;
 407   for(other = 0; other < size; other++) {
 408     if(other != rank) {
 409       requests[index] = smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm);
 410       index++;
 411       requests[index] = smpi_irecv_init(&((char*)recvbuf)[displs[other]], recvcounts[other], recvtype, other, system_tag, comm);
 412       index++;
 413     }
 414   }
 415   // Wait for completion of all comms.
 416   smpi_mpi_startall(2 * (size - 1), requests);
 417   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
 418   xbt_free(requests);
 419 }
 420
 421 void smpi_mpi_scatter(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) {
 422   int system_tag = 666;
 423   int rank, size, dst, index, sendsize, recvsize;
 424   MPI_Request* requests;
 425
 426   rank = smpi_comm_rank(comm);
 427   size = smpi_comm_size(comm);
 428   if(rank != root) {
 429     // Recv buffer from root
 430     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE);
 431   } else {
 432     sendsize = smpi_datatype_size(sendtype);
 433     recvsize = smpi_datatype_size(recvtype);
 434     // Local copy from root
 435     memcpy(recvbuf, &((char*)sendbuf)[root * sendcount * sendsize], recvcount * recvsize * sizeof(char));
 436     // Send buffers to receivers
 437     requests = xbt_new(MPI_Request, size - 1);
 438     index = 0;
 439     for(dst = 0; dst < size; dst++) {
 440       if(dst != root) {
 441         requests[index] = smpi_isend_init(&((char*)sendbuf)[dst * sendcount * sendsize], sendcount, sendtype, dst, system_tag, comm);
 442         index++;
 443       }
 444     }
 445     // Wait for completion of isend's.
 446     smpi_mpi_startall(size - 1, requests);
 447     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 448     xbt_free(requests);
 449   }
 450 }
 451
 452 void smpi_mpi_scatterv(void* sendbuf, int* sendcounts, int* displs, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) {
 453   int system_tag = 666;
 454   int rank, size, dst, index, sendsize, recvsize;
 455   MPI_Request* requests;
 456
 457   rank = smpi_comm_rank(comm);
 458   size = smpi_comm_size(comm);
 459   if(rank != root) {
 460     // Recv buffer from root
 461     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE);
 462   } else {
 463     sendsize = smpi_datatype_size(sendtype);
 464     recvsize = smpi_datatype_size(recvtype);
 465     // Local copy from root
 466     memcpy(recvbuf, &((char*)sendbuf)[displs[root]], recvcount * recvsize * sizeof(char));
 467     // Send buffers to receivers
 468     requests = xbt_new(MPI_Request, size - 1);
 469     index = 0;
 470     for(dst = 0; dst < size; dst++) {
 471       if(dst != root) {
 472         requests[index] = smpi_isend_init(&((char*)sendbuf)[displs[dst]], sendcounts[dst], sendtype, dst, system_tag, comm);
 473         index++;
 474       }
 475     }
 476     // Wait for completion of isend's.
 477     smpi_mpi_startall(size - 1, requests);
 478     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 479     xbt_free(requests);
 480   }
 481 }
 482
 483 void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, MPI_Comm comm) {
 484   int system_tag = 666;
 485   int rank, size, src, index, datasize;
 486   MPI_Request* requests;
 487   void** tmpbufs;
 488
 489   rank = smpi_comm_rank(comm);
 490   size = smpi_comm_size(comm);
 491   if(rank != root) {
 492     // Send buffer to root
 493     smpi_mpi_send(sendbuf, count, datatype, root, system_tag, comm);
 494   } else {
 495     datasize = smpi_datatype_size(datatype);
 496     // Local copy from root
 497     memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
 498     // Receive buffers from senders
 499     //TODO: make a MPI_barrier here ?
 500     requests = xbt_new(MPI_Request, size - 1);
 501     tmpbufs = xbt_new(void*, size - 1);
 502     index = 0;
 503     for(src = 0; src < size; src++) {
 504       if(src != root) {
 505         tmpbufs[index] = xbt_malloc(count * datasize);
 506         requests[index] = smpi_irecv_init(tmpbufs[index], count, datatype, src, system_tag, comm);
 507         index++;
 508       }
 509     }
 510     // Wait for completion of irecv's.
 511     smpi_mpi_startall(size - 1, requests);
 512     for(src = 0; src < size - 1; src++) {
 513       index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
 514       if(index == MPI_UNDEFINED) {
 515         break;
 516       }
 517       smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
 518     }
 519     for(index = 0; index < size - 1; index++) {
 520       xbt_free(tmpbufs[index]);
 521     }
 522     xbt_free(tmpbufs);
 523     xbt_free(requests);
 524   }
 525 }
 526
 527 void smpi_mpi_allreduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
 528   smpi_mpi_reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
 529   smpi_mpi_bcast(recvbuf, count, datatype, 0, comm);
 530
 531 /*
 532 FIXME: buggy implementation
 533
 534   int system_tag = 666;
 535   int rank, size, other, index, datasize;
 536   MPI_Request* requests;
 537   void** tmpbufs;
 538
 539   rank = smpi_comm_rank(comm);
 540   size = smpi_comm_size(comm);
 541   datasize = smpi_datatype_size(datatype);
 542   // Local copy from self
 543   memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
 544   // Send/Recv buffers to/from others;
 545   //TODO: make a MPI_barrier here ?
 546   requests = xbt_new(MPI_Request, 2 * (size - 1));
 547   tmpbufs = xbt_new(void*, size - 1);
 548   index = 0;
 549   for(other = 0; other < size; other++) {
 550     if(other != rank) {
 551       tmpbufs[index / 2] = xbt_malloc(count * datasize);
 552       requests[index] = smpi_mpi_isend(sendbuf, count, datatype, other, system_tag, comm);
 553       requests[index + 1] = smpi_mpi_irecv(tmpbufs[index / 2], count, datatype, other, system_tag, comm);
 554       index += 2;
 555     }
 556   }
 557   // Wait for completion of all comms.
 558   for(other = 0; other < 2 * (size - 1); other++) {
 559     index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
 560     if(index == MPI_UNDEFINED) {
 561       break;
 562     }
 563     if((index & 1) == 1) {
 564       // Request is odd: it's a irecv
 565       smpi_op_apply(op, tmpbufs[index / 2], recvbuf, &count, &datatype);
 566     }
 567   }
 568   for(index = 0; index < size - 1; index++) {
 569     xbt_free(tmpbufs[index]);
 570   }
 571   xbt_free(tmpbufs);
 572   xbt_free(requests);
 573 */
 574 }
 575
 576 void smpi_mpi_scan(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
 577   int system_tag = 666;
 578   int rank, size, other, index, datasize;
 579   int total;
 580   MPI_Request* requests;
 581   void** tmpbufs;
 582
 583   rank = smpi_comm_rank(comm);
 584   size = smpi_comm_size(comm);
 585   datasize = smpi_datatype_size(datatype);
 586   // Local copy from self
 587   memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
 588   // Send/Recv buffers to/from others;
 589   total = rank + (size - (rank + 1));
 590   requests = xbt_new(MPI_Request, total);
 591   tmpbufs = xbt_new(void*, rank);
 592   index = 0;
 593   for(other = 0; other < rank; other++) {
 594     tmpbufs[index] = xbt_malloc(count * datasize);
 595     requests[index] = smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag, comm);
 596     index++;
 597   }
 598   for(other = rank + 1; other < size; other++) {
 599     requests[index] = smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm);
 600     index++;
 601   }
 602   // Wait for completion of all comms.
 603   smpi_mpi_startall(size - 1, requests);
 604   for(other = 0; other < total; other++) {
 605     index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
 606     if(index == MPI_UNDEFINED) {
 607       break;
 608     }
 609     if(index < rank) {
 610       // #Request is below rank: it's a irecv
 611       smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
 612     }
 613   }
 614   for(index = 0; index < size - 1; index++) {
 615     xbt_free(tmpbufs[index]);
 616   }
 617   xbt_free(tmpbufs);
 618   xbt_free(requests);
 619 }