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