src/smpi/smpi_base.c

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