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 static int match_recv(void* a, void* b) {
  23    MPI_Request ref = (MPI_Request)a;
  24    MPI_Request req = (MPI_Request)b;
  25
  26    xbt_assert0(ref, "Cannot match recv against null reference");
  27    xbt_assert0(req, "Cannot match recv against null request");
  28    return req->comm == ref->comm
  29           && (ref->src == MPI_ANY_SOURCE || req->src == ref->src)
  30           && (ref->tag == MPI_ANY_TAG || req->tag == ref->tag);
  31 }
  32
  33 static int match_send(void* a, void* b) {
  34    MPI_Request ref = (MPI_Request)a;
  35    MPI_Request req = (MPI_Request)b;
  36
  37    xbt_assert0(ref, "Cannot match send against null reference");
  38    xbt_assert0(req, "Cannot match send against null request");
  39    return req->comm == ref->comm
  40           && (req->src == MPI_ANY_SOURCE || req->src == ref->src)
  41           && (req->tag == MPI_ANY_TAG || req->tag == ref->tag);
  42 }
  43
  44 static MPI_Request build_request(void *buf, int count,
  45                                  MPI_Datatype datatype, int src, int dst,
  46                                  int tag, MPI_Comm comm, unsigned flags)
  47 {
  48   MPI_Request request;
  49
  50   request = xbt_new(s_smpi_mpi_request_t, 1);
  51   request->buf = buf;
  52   request->size = smpi_datatype_size(datatype) * count;
  53   request->src = src;
  54   request->dst = dst;
  55   request->tag = tag;
  56   request->comm = comm;
  57   request->action = NULL;
  58   request->flags = flags;
  59 #ifdef HAVE_TRACING
  60   request->send = 0;
  61   request->recv = 0;
  62 #endif
  63   return request;
  64 }
  65
  66 /* MPI Low level calls */
  67 MPI_Request smpi_mpi_send_init(void *buf, int count, MPI_Datatype datatype,
  68                                int dst, int tag, MPI_Comm comm)
  69 {
  70   MPI_Request request =
  71       build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
  72                     comm, PERSISTENT | SEND);
  73
  74   return request;
  75 }
  76
  77 MPI_Request smpi_mpi_recv_init(void *buf, int count, MPI_Datatype datatype,
  78                                int src, int tag, MPI_Comm comm)
  79 {
  80   MPI_Request request =
  81       build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag,
  82                     comm, PERSISTENT | RECV);
  83
  84   return request;
  85 }
  86
  87 void smpi_mpi_start(MPI_Request request)
  88 {
  89   smx_rdv_t mailbox;
  90
  91   xbt_assert0(!request->action,
  92               "Cannot (re)start a non-finished communication");
  93   if(request->flags & RECV) {
  94     print_request("New recv", request);
  95     mailbox = smpi_process_mailbox();
  96     request->action = SIMIX_req_comm_irecv(mailbox, request->buf, &request->size, &match_recv, request);
  97   } else {
  98     print_request("New send", request);
  99     mailbox = smpi_process_remote_mailbox(request->dst);
 100     request->action = SIMIX_req_comm_isend(mailbox, request->size, -1.0,
 101                                            request->buf, request->size, &match_send, request);
 102 #ifdef HAVE_TRACING
 103     SIMIX_req_set_category (request->action, TRACE_internal_smpi_get_category());
 104 #endif
 105   }
 106 }
 107
 108 void smpi_mpi_startall(int count, MPI_Request * requests)
 109 {
 110   int i;
 111
 112   for(i = 0; i < count; i++) {
 113     smpi_mpi_start(requests[i]);
 114   }
 115 }
 116
 117 void smpi_mpi_request_free(MPI_Request * request)
 118 {
 119   xbt_free(*request);
 120   *request = MPI_REQUEST_NULL;
 121 }
 122
 123 MPI_Request smpi_isend_init(void *buf, int count, MPI_Datatype datatype,
 124                             int dst, int tag, MPI_Comm comm)
 125 {
 126   MPI_Request request =
 127       build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
 128                     comm, NON_PERSISTENT | SEND);
 129
 130   return request;
 131 }
 132
 133 MPI_Request smpi_mpi_isend(void *buf, int count, MPI_Datatype datatype,
 134                            int dst, int tag, MPI_Comm comm)
 135 {
 136   MPI_Request request =
 137       smpi_isend_init(buf, count, datatype, dst, tag, comm);
 138
 139   smpi_mpi_start(request);
 140   return request;
 141 }
 142
 143 MPI_Request smpi_irecv_init(void *buf, int count, MPI_Datatype datatype,
 144                             int src, int tag, MPI_Comm comm)
 145 {
 146   MPI_Request request =
 147       build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag,
 148                     comm, NON_PERSISTENT | RECV);
 149
 150   return request;
 151 }
 152
 153 MPI_Request smpi_mpi_irecv(void *buf, int count, MPI_Datatype datatype,
 154                            int src, int tag, MPI_Comm comm)
 155 {
 156   MPI_Request request =
 157       smpi_irecv_init(buf, count, datatype, src, tag, comm);
 158
 159   smpi_mpi_start(request);
 160   return request;
 161 }
 162
 163 void smpi_mpi_recv(void *buf, int count, MPI_Datatype datatype, int src,
 164                    int tag, MPI_Comm comm, MPI_Status * status)
 165 {
 166   MPI_Request request;
 167
 168   request = smpi_mpi_irecv(buf, count, datatype, src, tag, comm);
 169   smpi_mpi_wait(&request, status);
 170 }
 171
 172 void smpi_mpi_send(void *buf, int count, MPI_Datatype datatype, int dst,
 173                    int tag, MPI_Comm comm)
 174 {
 175   MPI_Request request;
 176
 177   request = smpi_mpi_isend(buf, count, datatype, dst, tag, comm);
 178   smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
 179 }
 180
 181 void smpi_mpi_sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 182                        int dst, int sendtag, void *recvbuf, int recvcount,
 183                        MPI_Datatype recvtype, int src, int recvtag,
 184                        MPI_Comm comm, MPI_Status * status)
 185 {
 186   MPI_Request requests[2];
 187   MPI_Status stats[2];
 188
 189   requests[0] =
 190       smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
 191   requests[1] =
 192       smpi_irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm);
 193   smpi_mpi_startall(2, requests);
 194   smpi_mpi_waitall(2, requests, stats);
 195   if(status != MPI_STATUS_IGNORE) {
 196     // Copy receive status
 197     memcpy(status, &stats[1], sizeof(MPI_Status));
 198   }
 199 }
 200
 201 int smpi_mpi_get_count(MPI_Status * status, MPI_Datatype datatype)
 202 {
 203   return status->count / smpi_datatype_size(datatype);
 204 }
 205
 206 static void finish_wait(MPI_Request * request, MPI_Status * status)
 207 {
 208   MPI_Request req = *request;
 209
 210   if(status != MPI_STATUS_IGNORE) {
 211     status->MPI_SOURCE = req->src;
 212     status->MPI_TAG = req->tag;
 213     status->MPI_ERROR = MPI_SUCCESS;
 214     status->count = SIMIX_req_comm_get_dst_buff_size(req->action);
 215   }
 216   SIMIX_req_comm_destroy(req->action);
 217   print_request("Finishing", req);
 218   if(req->flags & NON_PERSISTENT) {
 219     smpi_mpi_request_free(request);
 220   } else {
 221     req->action = NULL;
 222   }
 223 }
 224
 225 int smpi_mpi_test(MPI_Request * request, MPI_Status * status)
 226 {
 227    int flag = SIMIX_req_comm_test((*request)->action);
 228
 229    if(flag) {
 230       smpi_mpi_wait(request, status);
 231    }
 232    return flag;
 233 }
 234
 235 int smpi_mpi_testany(int count, MPI_Request requests[], int *index,
 236                      MPI_Status * status)
 237 {
 238   xbt_dynar_t comms;
 239   int i, flag, size;
 240   int* map;
 241
 242   *index = MPI_UNDEFINED;
 243   flag = 0;
 244   if(count > 0) {
 245     comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
 246     map = xbt_new(int, count);
 247     size = 0;
 248     for(i = 0; i < count; i++) {
 249       if(requests[i]->action) {
 250          xbt_dynar_push(comms, &requests[i]->action);
 251          map[size] = i;
 252          size++;
 253       }
 254     }
 255     if(size > 0) {
 256       *index = SIMIX_req_comm_testany(comms);
 257       *index = map[*index];
 258       if(*index != MPI_UNDEFINED) {
 259         smpi_mpi_wait(&requests[*index], status);
 260         flag = 1;
 261       }
 262     }
 263     xbt_free(map);
 264     xbt_dynar_free(&comms);
 265   }
 266   return flag;
 267 }
 268
 269 void smpi_mpi_wait(MPI_Request * request, MPI_Status * status)
 270 {
 271   print_request("Waiting", *request);
 272   SIMIX_req_comm_wait((*request)->action, -1.0);
 273   finish_wait(request, status);
 274 }
 275
 276 int smpi_mpi_waitany(int count, MPI_Request requests[],
 277                      MPI_Status * status)
 278 {
 279   xbt_dynar_t comms;
 280   int i, size, index;
 281   int *map;
 282
 283   index = MPI_UNDEFINED;
 284   if(count > 0) {
 285     // Wait for a request to complete
 286     comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
 287     map = xbt_new(int, count);
 288     size = 0;
 289     DEBUG0("Wait for one of");
 290     for(i = 0; i < count; i++) {
 291       if(requests[i] != MPI_REQUEST_NULL) {
 292         print_request("   ", requests[i]);
 293         xbt_dynar_push(comms, &requests[i]->action);
 294         map[size] = i;
 295         size++;
 296       }
 297     }
 298     if(size > 0) {
 299       index = SIMIX_req_comm_waitany(comms);
 300       index = map[index];
 301       finish_wait(&requests[index], status);
 302     }
 303     xbt_free(map);
 304     xbt_dynar_free(&comms);
 305   }
 306   return index;
 307 }
 308
 309 void smpi_mpi_waitall(int count, MPI_Request requests[],
 310                       MPI_Status status[])
 311 {
 312   int index, c;
 313   MPI_Status stat;
 314   MPI_Status *pstat = status == MPI_STATUS_IGNORE ? MPI_STATUS_IGNORE : &stat;
 315
 316   c = count;
 317   while(c > 0) {
 318     index = smpi_mpi_waitany(count, requests, pstat);
 319     if(index == MPI_UNDEFINED) {
 320       break;
 321     }
 322     if(status != MPI_STATUS_IGNORE) {
 323       memcpy(&status[index], pstat, sizeof *pstat);
 324     }
 325     c--;
 326   }
 327 }
 328
 329 int smpi_mpi_waitsome(int incount, MPI_Request requests[], int *indices,
 330                       MPI_Status status[])
 331 {
 332   int i, count, index;
 333
 334   count = 0;
 335   for(i = 0; i < incount; i++) {
 336      if(smpi_mpi_testany(incount, requests, &index, status)) {
 337        indices[count] = index;
 338        count++;
 339      }
 340   }
 341   return count;
 342 }
 343
 344 void smpi_mpi_bcast(void *buf, int count, MPI_Datatype datatype, int root,
 345                     MPI_Comm comm)
 346 {
 347   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
 348   nary_tree_bcast(buf, count, datatype, root, comm, 4);
 349 }
 350
 351 void smpi_mpi_barrier(MPI_Comm comm)
 352 {
 353   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
 354   nary_tree_barrier(comm, 4);
 355 }
 356
 357 void smpi_mpi_gather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 358                      void *recvbuf, int recvcount, MPI_Datatype recvtype,
 359                      int root, MPI_Comm comm)
 360 {
 361   int system_tag = 666;
 362   int rank, size, src, index, sendsize, recvsize;
 363   MPI_Request *requests;
 364
 365   rank = smpi_comm_rank(comm);
 366   size = smpi_comm_size(comm);
 367   if(rank != root) {
 368     // Send buffer to root
 369     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
 370   } else {
 371     sendsize = smpi_datatype_size(sendtype);
 372     recvsize = smpi_datatype_size(recvtype);
 373     // Local copy from root
 374     memcpy(&((char *) recvbuf)[root * recvcount * recvsize], sendbuf,
 375            sendcount * sendsize * sizeof(char));
 376     // Receive buffers from senders
 377     requests = xbt_new(MPI_Request, size - 1);
 378     index = 0;
 379     for(src = 0; src < size; src++) {
 380       if(src != root) {
 381         requests[index] = smpi_irecv_init(&((char *) recvbuf)
 382                                           [src * recvcount * recvsize],
 383                                           recvcount, recvtype, src,
 384                                           system_tag, comm);
 385         index++;
 386       }
 387     }
 388     // Wait for completion of irecv's.
 389     smpi_mpi_startall(size - 1, requests);
 390     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 391     xbt_free(requests);
 392   }
 393 }
 394
 395 void smpi_mpi_gatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 396                       void *recvbuf, int *recvcounts, int *displs,
 397                       MPI_Datatype recvtype, int root, MPI_Comm comm)
 398 {
 399   int system_tag = 666;
 400   int rank, size, src, index, sendsize;
 401   MPI_Request *requests;
 402
 403   rank = smpi_comm_rank(comm);
 404   size = smpi_comm_size(comm);
 405   if(rank != root) {
 406     // Send buffer to root
 407     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
 408   } else {
 409     sendsize = smpi_datatype_size(sendtype);
 410     // Local copy from root
 411     memcpy(&((char *) recvbuf)[displs[root]], sendbuf,
 412            sendcount * sendsize * sizeof(char));
 413     // Receive buffers from senders
 414     requests = xbt_new(MPI_Request, size - 1);
 415     index = 0;
 416     for(src = 0; src < size; src++) {
 417       if(src != root) {
 418         requests[index] =
 419             smpi_irecv_init(&((char *) recvbuf)[displs[src]],
 420                             recvcounts[src], recvtype, src, system_tag,
 421                             comm);
 422         index++;
 423       }
 424     }
 425     // Wait for completion of irecv's.
 426     smpi_mpi_startall(size - 1, requests);
 427     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 428     xbt_free(requests);
 429   }
 430 }
 431
 432 void smpi_mpi_allgather(void *sendbuf, int sendcount,
 433                         MPI_Datatype sendtype, void *recvbuf,
 434                         int recvcount, MPI_Datatype recvtype,
 435                         MPI_Comm comm)
 436 {
 437   int system_tag = 666;
 438   int rank, size, other, index, sendsize, recvsize;
 439   MPI_Request *requests;
 440
 441   rank = smpi_comm_rank(comm);
 442   size = smpi_comm_size(comm);
 443   sendsize = smpi_datatype_size(sendtype);
 444   recvsize = smpi_datatype_size(recvtype);
 445   // Local copy from self
 446   memcpy(&((char *) recvbuf)[rank * recvcount * recvsize], sendbuf,
 447          sendcount * sendsize * sizeof(char));
 448   // Send/Recv buffers to/from others;
 449   requests = xbt_new(MPI_Request, 2 * (size - 1));
 450   index = 0;
 451   for(other = 0; other < size; other++) {
 452     if(other != rank) {
 453       requests[index] =
 454           smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag,
 455                           comm);
 456       index++;
 457       requests[index] = smpi_irecv_init(&((char *) recvbuf)
 458                                         [other * recvcount * recvsize],
 459                                         recvcount, recvtype, other,
 460                                         system_tag, comm);
 461       index++;
 462     }
 463   }
 464   // Wait for completion of all comms.
 465   smpi_mpi_startall(2 * (size - 1), requests);
 466   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
 467   xbt_free(requests);
 468 }
 469
 470 void smpi_mpi_allgatherv(void *sendbuf, int sendcount,
 471                          MPI_Datatype sendtype, void *recvbuf,
 472                          int *recvcounts, int *displs,
 473                          MPI_Datatype recvtype, MPI_Comm comm)
 474 {
 475   int system_tag = 666;
 476   int rank, size, other, index, sendsize, recvsize;
 477   MPI_Request *requests;
 478
 479   rank = smpi_comm_rank(comm);
 480   size = smpi_comm_size(comm);
 481   sendsize = smpi_datatype_size(sendtype);
 482   recvsize = smpi_datatype_size(recvtype);
 483   // Local copy from self
 484   memcpy(&((char *) recvbuf)[displs[rank]], sendbuf,
 485          sendcount * sendsize * sizeof(char));
 486   // Send buffers to others;
 487   requests = xbt_new(MPI_Request, 2 * (size - 1));
 488   index = 0;
 489   for(other = 0; other < size; other++) {
 490     if(other != rank) {
 491       requests[index] =
 492           smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag,
 493                           comm);
 494       index++;
 495       requests[index] =
 496           smpi_irecv_init(&((char *) recvbuf)[displs[other]],
 497                           recvcounts[other], recvtype, other, system_tag,
 498                           comm);
 499       index++;
 500     }
 501   }
 502   // Wait for completion of all comms.
 503   smpi_mpi_startall(2 * (size - 1), requests);
 504   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
 505   xbt_free(requests);
 506 }
 507
 508 void smpi_mpi_scatter(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 509                       void *recvbuf, int recvcount, MPI_Datatype recvtype,
 510                       int root, MPI_Comm comm)
 511 {
 512   int system_tag = 666;
 513   int rank, size, dst, index, sendsize, recvsize;
 514   MPI_Request *requests;
 515
 516   rank = smpi_comm_rank(comm);
 517   size = smpi_comm_size(comm);
 518   if(rank != root) {
 519     // Recv buffer from root
 520     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm,
 521                   MPI_STATUS_IGNORE);
 522   } else {
 523     sendsize = smpi_datatype_size(sendtype);
 524     recvsize = smpi_datatype_size(recvtype);
 525     // Local copy from root
 526     memcpy(recvbuf, &((char *) sendbuf)[root * sendcount * sendsize],
 527            recvcount * recvsize * sizeof(char));
 528     // Send buffers to receivers
 529     requests = xbt_new(MPI_Request, size - 1);
 530     index = 0;
 531     for(dst = 0; dst < size; dst++) {
 532       if(dst != root) {
 533         requests[index] = smpi_isend_init(&((char *) sendbuf)
 534                                           [dst * sendcount * sendsize],
 535                                           sendcount, sendtype, dst,
 536                                           system_tag, comm);
 537         index++;
 538       }
 539     }
 540     // Wait for completion of isend's.
 541     smpi_mpi_startall(size - 1, requests);
 542     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 543     xbt_free(requests);
 544   }
 545 }
 546
 547 void smpi_mpi_scatterv(void *sendbuf, int *sendcounts, int *displs,
 548                        MPI_Datatype sendtype, void *recvbuf, int recvcount,
 549                        MPI_Datatype recvtype, int root, MPI_Comm comm)
 550 {
 551   int system_tag = 666;
 552   int rank, size, dst, index, sendsize, recvsize;
 553   MPI_Request *requests;
 554
 555   rank = smpi_comm_rank(comm);
 556   size = smpi_comm_size(comm);
 557   if(rank != root) {
 558     // Recv buffer from root
 559     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm,
 560                   MPI_STATUS_IGNORE);
 561   } else {
 562     sendsize = smpi_datatype_size(sendtype);
 563     recvsize = smpi_datatype_size(recvtype);
 564     // Local copy from root
 565     memcpy(recvbuf, &((char *) sendbuf)[displs[root]],
 566            recvcount * recvsize * sizeof(char));
 567     // Send buffers to receivers
 568     requests = xbt_new(MPI_Request, size - 1);
 569     index = 0;
 570     for(dst = 0; dst < size; dst++) {
 571       if(dst != root) {
 572         requests[index] =
 573             smpi_isend_init(&((char *) sendbuf)[displs[dst]],
 574                             sendcounts[dst], sendtype, dst, system_tag,
 575                             comm);
 576         index++;
 577       }
 578     }
 579     // Wait for completion of isend's.
 580     smpi_mpi_startall(size - 1, requests);
 581     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 582     xbt_free(requests);
 583   }
 584 }
 585
 586 void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count,
 587                      MPI_Datatype datatype, MPI_Op op, int root,
 588                      MPI_Comm comm)
 589 {
 590   int system_tag = 666;
 591   int rank, size, src, index, datasize;
 592   MPI_Request *requests;
 593   void **tmpbufs;
 594
 595   rank = smpi_comm_rank(comm);
 596   size = smpi_comm_size(comm);
 597   if(rank != root) {
 598     // Send buffer to root
 599     smpi_mpi_send(sendbuf, count, datatype, root, system_tag, comm);
 600   } else {
 601     datasize = smpi_datatype_size(datatype);
 602     // Local copy from root
 603     memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
 604     // Receive buffers from senders
 605     //TODO: make a MPI_barrier here ?
 606     requests = xbt_new(MPI_Request, size - 1);
 607     tmpbufs = xbt_new(void *, size - 1);
 608     index = 0;
 609     for(src = 0; src < size; src++) {
 610       if(src != root) {
 611         tmpbufs[index] = xbt_malloc(count * datasize);
 612         requests[index] =
 613             smpi_irecv_init(tmpbufs[index], count, datatype, src,
 614                             system_tag, comm);
 615         index++;
 616       }
 617     }
 618     // Wait for completion of irecv's.
 619     smpi_mpi_startall(size - 1, requests);
 620     for(src = 0; src < size - 1; src++) {
 621       index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
 622       if(index == MPI_UNDEFINED) {
 623         break;
 624       }
 625       smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
 626     }
 627     for(index = 0; index < size - 1; index++) {
 628       xbt_free(tmpbufs[index]);
 629     }
 630     xbt_free(tmpbufs);
 631     xbt_free(requests);
 632   }
 633 }
 634
 635 void smpi_mpi_allreduce(void *sendbuf, void *recvbuf, int count,
 636                         MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
 637 {
 638   smpi_mpi_reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
 639   smpi_mpi_bcast(recvbuf, count, datatype, 0, comm);
 640
 641 /*
 642 FIXME: buggy implementation
 643
 644   int system_tag = 666;
 645   int rank, size, other, index, datasize;
 646   MPI_Request* requests;
 647   void** tmpbufs;
 648
 649   rank = smpi_comm_rank(comm);
 650   size = smpi_comm_size(comm);
 651   datasize = smpi_datatype_size(datatype);
 652   // Local copy from self
 653   memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
 654   // Send/Recv buffers to/from others;
 655   //TODO: make a MPI_barrier here ?
 656   requests = xbt_new(MPI_Request, 2 * (size - 1));
 657   tmpbufs = xbt_new(void*, size - 1);
 658   index = 0;
 659   for(other = 0; other < size; other++) {
 660     if(other != rank) {
 661       tmpbufs[index / 2] = xbt_malloc(count * datasize);
 662       requests[index] = smpi_mpi_isend(sendbuf, count, datatype, other, system_tag, comm);
 663       requests[index + 1] = smpi_mpi_irecv(tmpbufs[index / 2], count, datatype, other, system_tag, comm);
 664       index += 2;
 665     }
 666   }
 667   // Wait for completion of all comms.
 668   for(other = 0; other < 2 * (size - 1); other++) {
 669     index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
 670     if(index == MPI_UNDEFINED) {
 671       break;
 672     }
 673     if((index & 1) == 1) {
 674       // Request is odd: it's a irecv
 675       smpi_op_apply(op, tmpbufs[index / 2], recvbuf, &count, &datatype);
 676     }
 677   }
 678   for(index = 0; index < size - 1; index++) {
 679     xbt_free(tmpbufs[index]);
 680   }
 681   xbt_free(tmpbufs);
 682   xbt_free(requests);
 683 */
 684 }
 685
 686 void smpi_mpi_scan(void *sendbuf, void *recvbuf, int count,
 687                    MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
 688 {
 689   int system_tag = 666;
 690   int rank, size, other, index, datasize;
 691   int total;
 692   MPI_Request *requests;
 693   void **tmpbufs;
 694
 695   rank = smpi_comm_rank(comm);
 696   size = smpi_comm_size(comm);
 697   datasize = smpi_datatype_size(datatype);
 698   // Local copy from self
 699   memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
 700   // Send/Recv buffers to/from others;
 701   total = rank + (size - (rank + 1));
 702   requests = xbt_new(MPI_Request, total);
 703   tmpbufs = xbt_new(void *, rank);
 704   index = 0;
 705   for(other = 0; other < rank; other++) {
 706     tmpbufs[index] = xbt_malloc(count * datasize);
 707     requests[index] =
 708         smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag,
 709                         comm);
 710     index++;
 711   }
 712   for(other = rank + 1; other < size; other++) {
 713     requests[index] =
 714         smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm);
 715     index++;
 716   }
 717   // Wait for completion of all comms.
 718   smpi_mpi_startall(size - 1, requests);
 719   for(other = 0; other < total; other++) {
 720     index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
 721     if(index == MPI_UNDEFINED) {
 722       break;
 723     }
 724     if(index < rank) {
 725       // #Request is below rank: it's a irecv
 726       smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
 727     }
 728   }
 729   for(index = 0; index < size - 1; index++) {
 730     xbt_free(tmpbufs[index]);
 731   }
 732   xbt_free(tmpbufs);
 733   xbt_free(requests);
 734 }