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/virtu.h"
   9 #include "mc/mc.h"
  10 #include "xbt/replay.h"
  11 #include <errno.h>
  12 #include "simix/smx_private.h"
  13 #include "surf/surf.h"
  14 #include "simgrid/sg_config.h"
  15
  16
  17 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)");
  18
  19
  20 static int match_recv(void* a, void* b, smx_action_t ignored) {
  21    MPI_Request ref = (MPI_Request)a;
  22    MPI_Request req = (MPI_Request)b;
  23    XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d",ref->src,req->src, ref->tag, req->tag);
  24
  25   xbt_assert(ref, "Cannot match recv against null reference");
  26   xbt_assert(req, "Cannot match recv against null request");
  27   if((ref->src == MPI_ANY_SOURCE || req->src == ref->src)
  28     && (ref->tag == MPI_ANY_TAG || req->tag == ref->tag)){
  29     //we match, we can transfer some values
  30     // FIXME : move this to the copy function ?
  31     if(ref->src == MPI_ANY_SOURCE)ref->real_src = req->src;
  32     if(ref->tag == MPI_ANY_TAG)ref->real_tag = req->tag;
  33     if(ref->real_size < req->real_size) ref->truncated = 1;
  34     if(req->detached==1){
  35         ref->detached_sender=req; //tie the sender to the receiver, as it is detached and has to be freed in the receiver
  36     }
  37     return 1;
  38   }else return 0;
  39 }
  40
  41 static int match_send(void* a, void* b,smx_action_t ignored) {
  42    MPI_Request ref = (MPI_Request)a;
  43    MPI_Request req = (MPI_Request)b;
  44    XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d",ref->src,req->src, ref->tag, req->tag);
  45    xbt_assert(ref, "Cannot match send against null reference");
  46    xbt_assert(req, "Cannot match send against null request");
  47
  48    if((req->src == MPI_ANY_SOURCE || req->src == ref->src)
  49              && (req->tag == MPI_ANY_TAG || req->tag == ref->tag))
  50    {
  51      if(req->src == MPI_ANY_SOURCE)req->real_src = ref->src;
  52      if(req->tag == MPI_ANY_TAG)req->real_tag = ref->tag;
  53      if(req->real_size < ref->real_size) req->truncated = 1;
  54      if(ref->detached==1){
  55          req->detached_sender=ref; //tie the sender to the receiver, as it is detached and has to be freed in the receiver
  56      }
  57
  58      return 1;
  59    } else return 0;
  60 }
  61
  62
  63 typedef struct s_smpi_factor *smpi_factor_t;
  64 typedef struct s_smpi_factor {
  65   long factor;
  66   int nb_values;
  67   double values[4];//arbitrary set to 4
  68 } s_smpi_factor_t;
  69 xbt_dynar_t smpi_os_values = NULL;
  70 xbt_dynar_t smpi_or_values = NULL;
  71
  72
  73 // Methods used to parse and store the values for timing injections in smpi
  74 // These are taken from surf/network.c and generalized to have more factors
  75 // These methods should be merged with those in surf/network.c (moved somewhere in xbt ?)
  76
  77 static int factor_cmp(const void *pa, const void *pb)
  78 {
  79   return (((s_smpi_factor_t*)pa)->factor > ((s_smpi_factor_t*)pb)->factor);
  80 }
  81
  82
  83 static xbt_dynar_t parse_factor(const char *smpi_coef_string)
  84 {
  85   char *value = NULL;
  86   unsigned int iter = 0;
  87   s_smpi_factor_t fact;
  88   int i=0;
  89   xbt_dynar_t smpi_factor, radical_elements, radical_elements2 = NULL;
  90
  91   smpi_factor = xbt_dynar_new(sizeof(s_smpi_factor_t), NULL);
  92   radical_elements = xbt_str_split(smpi_coef_string, ";");
  93   xbt_dynar_foreach(radical_elements, iter, value) {
  94     fact.nb_values=0;
  95     radical_elements2 = xbt_str_split(value, ":");
  96     if (xbt_dynar_length(radical_elements2) <2 || xbt_dynar_length(radical_elements2) > 5)
  97       xbt_die("Malformed radical for smpi factor!");
  98     for(i =0; i<xbt_dynar_length(radical_elements2);i++ ){
  99         if (i==0){
 100            fact.factor = atol(xbt_dynar_get_as(radical_elements2, i, char *));
 101         }else{
 102            fact.values[fact.nb_values] = atof(xbt_dynar_get_as(radical_elements2, i, char *));
 103            fact.nb_values++;
 104         }
 105     }
 106
 107     xbt_dynar_push_as(smpi_factor, s_smpi_factor_t, fact);
 108     XBT_DEBUG("smpi_factor:\t%ld : %d values, first: %f", fact.factor, fact.nb_values ,fact.values[0]);
 109     xbt_dynar_free(&radical_elements2);
 110   }
 111   xbt_dynar_free(&radical_elements);
 112   iter=0;
 113   xbt_dynar_sort(smpi_factor, &factor_cmp);
 114   xbt_dynar_foreach(smpi_factor, iter, fact) {
 115     XBT_DEBUG("smpi_factor:\t%ld : %d values, first: %f", fact.factor, fact.nb_values ,fact.values[0]);
 116   }
 117   return smpi_factor;
 118 }
 119
 120 static double smpi_os(double size)
 121 {
 122   if (!smpi_os_values)
 123     smpi_os_values =
 124         parse_factor(sg_cfg_get_string("smpi/os"));
 125
 126   unsigned int iter = 0;
 127   s_smpi_factor_t fact;
 128   double current=0.0;
 129   xbt_dynar_foreach(smpi_os_values, iter, fact) {
 130     if (size <= fact.factor) {
 131         XBT_DEBUG("os : %lf <= %ld return %f", size, fact.factor, current);
 132       return current;
 133     }else{
 134       current=fact.values[0]+fact.values[1]*size;
 135     }
 136   }
 137   XBT_DEBUG("os : %lf > %ld return %f", size, fact.factor, current);
 138
 139   return current;
 140 }
 141
 142 static double smpi_or(double size)
 143 {
 144   if (!smpi_or_values)
 145     smpi_or_values =
 146         parse_factor(sg_cfg_get_string("smpi/or"));
 147
 148   unsigned int iter = 0;
 149   s_smpi_factor_t fact;
 150   double current=0.0;
 151   xbt_dynar_foreach(smpi_or_values, iter, fact) {
 152     if (size <= fact.factor) {
 153         XBT_DEBUG("or : %lf <= %ld return %f", size, fact.factor, current);
 154       return current;
 155     }else
 156       current=fact.values[0]+fact.values[1]*size;
 157   }
 158   XBT_DEBUG("or : %lf > %ld return %f", size, fact.factor, current);
 159
 160   return current;
 161 }
 162
 163 static MPI_Request build_request(void *buf, int count,
 164                                  MPI_Datatype datatype, int src, int dst,
 165                                  int tag, MPI_Comm comm, unsigned flags)
 166 {
 167   MPI_Request request;
 168
 169   void *old_buf = NULL;
 170
 171   request = xbt_new(s_smpi_mpi_request_t, 1);
 172
 173   s_smpi_subtype_t *subtype = datatype->substruct;
 174
 175   if(datatype->has_subtype == 1){
 176     // This part handles the problem of non-contiguous memory
 177     old_buf = buf;
 178     buf = xbt_malloc(count*smpi_datatype_size(datatype));
 179     if (flags & SEND) {
 180       subtype->serialize(old_buf, buf, count, datatype->substruct);
 181     }
 182   }
 183
 184   request->buf = buf;
 185   // This part handles the problem of non-contiguous memory (for the
 186   // unserialisation at the reception)
 187   request->old_buf = old_buf;
 188   request->old_type = datatype;
 189
 190   request->size = smpi_datatype_size(datatype) * count;
 191   request->src = src;
 192   request->dst = dst;
 193   request->tag = tag;
 194   request->comm = comm;
 195   request->action = NULL;
 196   request->flags = flags;
 197   request->detached = 0;
 198   request->detached_sender = NULL;
 199
 200   request->truncated = 0;
 201   request->real_size = 0;
 202   request->real_tag = 0;
 203
 204   request->refcount=1;
 205 #ifdef HAVE_TRACING
 206   request->send = 0;
 207   request->recv = 0;
 208 #endif
 209   if (flags & SEND) smpi_datatype_unuse(datatype);
 210
 211   return request;
 212 }
 213
 214
 215 void smpi_empty_status(MPI_Status * status) {
 216   if(status != MPI_STATUS_IGNORE) {
 217       status->MPI_SOURCE=MPI_ANY_SOURCE;
 218       status->MPI_TAG=MPI_ANY_TAG;
 219       status->count=0;
 220   }
 221 }
 222
 223 void smpi_action_trace_run(char *path)
 224 {
 225   char *name;
 226   xbt_dynar_t todo;
 227   xbt_dict_cursor_t cursor;
 228
 229   action_fp=NULL;
 230   if (path) {
 231     action_fp = fopen(path, "r");
 232     xbt_assert(action_fp != NULL, "Cannot open %s: %s", path,
 233                strerror(errno));
 234   }
 235
 236   if (!xbt_dict_is_empty(action_queues)) {
 237     XBT_WARN
 238       ("Not all actions got consumed. If the simulation ended successfully (without deadlock), you may want to add new processes to your deployment file.");
 239
 240
 241     xbt_dict_foreach(action_queues, cursor, name, todo) {
 242       XBT_WARN("Still %lu actions for %s", xbt_dynar_length(todo), name);
 243     }
 244   }
 245
 246   if (path)
 247     fclose(action_fp);
 248   xbt_dict_free(&action_queues);
 249   action_queues = xbt_dict_new_homogeneous(NULL);
 250 }
 251
 252 static void smpi_mpi_request_free_voidp(void* request)
 253 {
 254   MPI_Request req = request;
 255   smpi_mpi_request_free(&req);
 256 }
 257
 258 /* MPI Low level calls */
 259 MPI_Request smpi_mpi_send_init(void *buf, int count, MPI_Datatype datatype,
 260                                int dst, int tag, MPI_Comm comm)
 261 {
 262   MPI_Request request =
 263     build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
 264                   comm, PERSISTENT | SEND);
 265   request->refcount++;
 266   return request;
 267 }
 268
 269 MPI_Request smpi_mpi_ssend_init(void *buf, int count, MPI_Datatype datatype,
 270                                int dst, int tag, MPI_Comm comm)
 271 {
 272   MPI_Request request =
 273     build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
 274                   comm, PERSISTENT | SSEND | SEND);
 275   request->refcount++;
 276   return request;
 277 }
 278
 279 MPI_Request smpi_mpi_recv_init(void *buf, int count, MPI_Datatype datatype,
 280                                int src, int tag, MPI_Comm comm)
 281 {
 282   MPI_Request request =
 283     build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag,
 284                   comm, PERSISTENT | RECV);
 285   request->refcount++;
 286   return request;
 287 }
 288
 289 void smpi_mpi_start(MPI_Request request)
 290 {
 291   smx_rdv_t mailbox;
 292
 293   xbt_assert(!request->action,
 294              "Cannot (re)start a non-finished communication");
 295   if(request->flags & RECV) {
 296     print_request("New recv", request);
 297     if (request->size < sg_cfg_get_int("smpi/async_small_thres"))
 298       mailbox = smpi_process_mailbox_small();
 299     else
 300       mailbox = smpi_process_mailbox();
 301     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
 302     request->real_size=request->size;
 303     smpi_datatype_use(request->old_type);
 304     request->action = simcall_comm_irecv(mailbox, request->buf, &request->real_size, &match_recv, request);
 305
 306     double sleeptime = smpi_or(request->size);
 307     if(sleeptime!=0.0){
 308         simcall_process_sleep(sleeptime);
 309         XBT_DEBUG("receiving size of %zu : sleep %lf ", request->size, smpi_or(request->size));
 310     }
 311
 312   } else {
 313
 314     int receiver = smpi_group_index(smpi_comm_group(request->comm), request->dst);
 315 /*    if(receiver == MPI_UNDEFINED) {*/
 316 /*      XBT_WARN("Trying to send a message to a wrong rank");*/
 317 /*      return;*/
 318 /*    }*/
 319     print_request("New send", request);
 320     if (request->size < sg_cfg_get_int("smpi/async_small_thres")) { // eager mode
 321       mailbox = smpi_process_remote_mailbox_small(receiver);
 322     }else{
 323       XBT_DEBUG("Send request %p is not in the permanent receive mailbox (buf: %p)",request,request->buf);
 324       mailbox = smpi_process_remote_mailbox(receiver);
 325     }
 326     if ( (! (request->flags & SSEND)) && (request->size < sg_cfg_get_int("smpi/send_is_detached_thres"))) {
 327       void *oldbuf = NULL;
 328       request->detached = 1;
 329       request->refcount++;
 330       if(request->old_type->has_subtype == 0){
 331         oldbuf = request->buf;
 332         if (oldbuf){
 333           request->buf = xbt_malloc(request->size);
 334           memcpy(request->buf,oldbuf,request->size);
 335         }
 336       }
 337       XBT_DEBUG("Send request %p is detached; buf %p copied into %p",request,oldbuf,request->buf);
 338     }
 339     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
 340     request->real_size=request->size;
 341     smpi_datatype_use(request->old_type);
 342     double sleeptime = smpi_os(request->size);
 343     if(sleeptime!=0.0){
 344         simcall_process_sleep(sleeptime);
 345         XBT_DEBUG("sending size of %zu : sleep %lf ", request->size, smpi_os(request->size));
 346     }
 347     request->action =
 348       simcall_comm_isend(mailbox, request->size, -1.0,
 349                          request->buf, request->real_size,
 350                          &match_send,
 351                          &smpi_mpi_request_free_voidp, // how to free the userdata if a detached send fails
 352                          request,
 353                          // detach if msg size < eager/rdv switch limit
 354                          request->detached);
 355
 356 #ifdef HAVE_TRACING
 357     /* FIXME: detached sends are not traceable (request->action == NULL) */
 358     if (request->action)
 359       simcall_set_category(request->action, TRACE_internal_smpi_get_category());
 360 #endif
 361
 362   }
 363
 364 }
 365
 366 void smpi_mpi_startall(int count, MPI_Request * requests)
 367 {
 368   int i;
 369
 370   for(i = 0; i < count; i++) {
 371     smpi_mpi_start(requests[i]);
 372   }
 373 }
 374
 375 void smpi_mpi_request_free(MPI_Request * request)
 376 {
 377
 378   if((*request) != MPI_REQUEST_NULL){
 379     (*request)->refcount--;
 380     if((*request)->refcount<0) xbt_die("wrong refcount");
 381
 382     if((*request)->refcount==0){
 383         xbt_free(*request);
 384         *request = MPI_REQUEST_NULL;
 385     }
 386   }else{
 387       xbt_die("freeing an already free request");
 388   }
 389 }
 390
 391 MPI_Request smpi_isend_init(void *buf, int count, MPI_Datatype datatype,
 392                             int dst, int tag, MPI_Comm comm)
 393 {
 394   MPI_Request request =
 395     build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
 396                   comm, NON_PERSISTENT | SEND);
 397
 398   return request;
 399 }
 400
 401 MPI_Request smpi_mpi_isend(void *buf, int count, MPI_Datatype datatype,
 402                            int dst, int tag, MPI_Comm comm)
 403 {
 404   MPI_Request request =
 405       build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
 406                     comm, NON_PERSISTENT | ISEND | SEND);
 407
 408   smpi_mpi_start(request);
 409   return request;
 410 }
 411
 412 MPI_Request smpi_mpi_issend(void *buf, int count, MPI_Datatype datatype,
 413                            int dst, int tag, MPI_Comm comm)
 414 {
 415   MPI_Request request =
 416       build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
 417                     comm, NON_PERSISTENT | ISEND | SSEND | SEND);
 418   smpi_mpi_start(request);
 419   return request;
 420 }
 421
 422
 423
 424 MPI_Request smpi_irecv_init(void *buf, int count, MPI_Datatype datatype,
 425                             int src, int tag, MPI_Comm comm)
 426 {
 427   MPI_Request request =
 428     build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag,
 429                   comm, NON_PERSISTENT | RECV);
 430   return request;
 431 }
 432
 433 MPI_Request smpi_mpi_irecv(void *buf, int count, MPI_Datatype datatype,
 434                            int src, int tag, MPI_Comm comm)
 435 {
 436   MPI_Request request =
 437       build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag,
 438                     comm, NON_PERSISTENT | RECV);
 439
 440   smpi_mpi_start(request);
 441   return request;
 442 }
 443
 444 void smpi_mpi_recv(void *buf, int count, MPI_Datatype datatype, int src,
 445                    int tag, MPI_Comm comm, MPI_Status * status)
 446 {
 447   MPI_Request request;
 448   request = smpi_mpi_irecv(buf, count, datatype, src, tag, comm);
 449   smpi_mpi_wait(&request, status);
 450 }
 451
 452
 453
 454 void smpi_mpi_send(void *buf, int count, MPI_Datatype datatype, int dst,
 455                    int tag, MPI_Comm comm)
 456 {
 457   MPI_Request request;
 458   request = smpi_mpi_isend(buf, count, datatype, dst, tag, comm);
 459   smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
 460
 461 }
 462
 463 void smpi_mpi_ssend(void *buf, int count, MPI_Datatype datatype,
 464                            int dst, int tag, MPI_Comm comm)
 465 {
 466   MPI_Request request = smpi_mpi_issend(buf, count, datatype, dst, tag, comm);
 467   smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
 468 }
 469
 470 void smpi_mpi_sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 471                        int dst, int sendtag, void *recvbuf, int recvcount,
 472                        MPI_Datatype recvtype, int src, int recvtag,
 473                        MPI_Comm comm, MPI_Status * status)
 474 {
 475   MPI_Request requests[2];
 476   MPI_Status stats[2];
 477
 478   requests[0] =
 479     smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
 480   requests[1] =
 481     smpi_irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm);
 482   smpi_mpi_startall(2, requests);
 483   smpi_mpi_waitall(2, requests, stats);
 484   if(status != MPI_STATUS_IGNORE) {
 485     // Copy receive status
 486     memcpy(status, &stats[1], sizeof(MPI_Status));
 487   }
 488 }
 489
 490 int smpi_mpi_get_count(MPI_Status * status, MPI_Datatype datatype)
 491 {
 492   return status->count / smpi_datatype_size(datatype);
 493 }
 494
 495 static void finish_wait(MPI_Request * request, MPI_Status * status)
 496 {
 497   MPI_Request req = *request;
 498   if(!(req->detached && req->flags & SEND)){
 499     if(status != MPI_STATUS_IGNORE) {
 500       status->MPI_SOURCE = req->src == MPI_ANY_SOURCE ? req->real_src : req->src;
 501       status->MPI_TAG = req->tag == MPI_ANY_TAG ? req->real_tag : req->tag;
 502       if(req->truncated)
 503       status->MPI_ERROR = MPI_ERR_TRUNCATE;
 504       else status->MPI_ERROR = MPI_SUCCESS ;
 505       // this handles the case were size in receive differs from size in send
 506       // FIXME: really this should just contain the count of receive-type blocks,
 507       // right?
 508       status->count = req->real_size;
 509     }
 510
 511     print_request("Finishing", req);
 512     MPI_Datatype datatype = req->old_type;
 513
 514     if(datatype->has_subtype == 1){
 515         // This part handles the problem of non-contignous memory
 516         // the unserialization at the reception
 517       s_smpi_subtype_t *subtype = datatype->substruct;
 518       if(req->flags & RECV) {
 519         subtype->unserialize(req->buf, req->old_buf, req->real_size/smpi_datatype_size(datatype) , datatype->substruct);
 520       }
 521       if(req->detached == 0) free(req->buf);
 522     }
 523     smpi_datatype_unuse(datatype);
 524   }
 525
 526   if(req->detached_sender!=NULL){
 527     smpi_mpi_request_free(&(req->detached_sender));
 528   }
 529
 530   if(req->flags & NON_PERSISTENT) {
 531     smpi_mpi_request_free(request);
 532   } else {
 533     req->action = NULL;
 534   }
 535 }
 536
 537 int smpi_mpi_test(MPI_Request * request, MPI_Status * status) {
 538   int flag;
 539
 540   //assume that request is not MPI_REQUEST_NULL (filtered in PMPI_Test or smpi_mpi_testall before)
 541   if ((*request)->action == NULL)
 542     flag = 1;
 543   else
 544     flag = simcall_comm_test((*request)->action);
 545   if(flag) {
 546     (*request)->refcount++;
 547     finish_wait(request, status);
 548   }else{
 549     smpi_empty_status(status);
 550   }
 551   return flag;
 552 }
 553
 554 int smpi_mpi_testany(int count, MPI_Request requests[], int *index,
 555                      MPI_Status * status)
 556 {
 557   xbt_dynar_t comms;
 558   int i, flag, size;
 559   int* map;
 560
 561   *index = MPI_UNDEFINED;
 562   flag = 0;
 563   if(count > 0) {
 564     comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
 565     map = xbt_new(int, count);
 566     size = 0;
 567     for(i = 0; i < count; i++) {
 568       if((requests[i]!=MPI_REQUEST_NULL) && requests[i]->action) {
 569          xbt_dynar_push(comms, &requests[i]->action);
 570          map[size] = i;
 571          size++;
 572       }
 573     }
 574     if(size > 0) {
 575       i = simcall_comm_testany(comms);
 576       // not MPI_UNDEFINED, as this is a simix return code
 577       if(i != -1) {
 578         *index = map[i];
 579         finish_wait(&requests[*index], status);
 580         flag = 1;
 581       }
 582     }else{
 583         //all requests are null or inactive, return true
 584         flag=1;
 585         smpi_empty_status(status);
 586     }
 587     xbt_free(map);
 588     xbt_dynar_free(&comms);
 589   }
 590
 591   return flag;
 592 }
 593
 594
 595 int smpi_mpi_testall(int count, MPI_Request requests[],
 596                      MPI_Status status[])
 597 {
 598   MPI_Status stat;
 599   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 600   int flag=1;
 601   int i;
 602   for(i=0; i<count; i++){
 603     if(requests[i]!= MPI_REQUEST_NULL){
 604       if (smpi_mpi_test(&requests[i], pstat)!=1){
 605         flag=0;
 606       }
 607     }else{
 608       smpi_empty_status(pstat);
 609     }
 610     if(status != MPI_STATUSES_IGNORE) {
 611       memcpy(&status[i], pstat, sizeof(*pstat));
 612     }
 613   }
 614   return flag;
 615 }
 616
 617 void smpi_mpi_probe(int source, int tag, MPI_Comm comm, MPI_Status* status){
 618   int flag=0;
 619   //FIXME find another wait to avoid busy waiting ?
 620   // the issue here is that we have to wait on a nonexistent comm
 621   while(flag==0){
 622     smpi_mpi_iprobe(source, tag, comm, &flag, status);
 623     XBT_DEBUG("Busy Waiting on probing : %d", flag);
 624     if(!flag) {
 625       simcall_process_sleep(0.0001);
 626     }
 627   }
 628 }
 629
 630 void smpi_mpi_iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* status){
 631   MPI_Request request =build_request(NULL, 0, MPI_CHAR, source, smpi_comm_rank(comm), tag,
 632             comm, NON_PERSISTENT | RECV);
 633
 634   // behave like a receive, but don't do it
 635   smx_rdv_t mailbox;
 636
 637   print_request("New iprobe", request);
 638   // We have to test both mailboxes as we don't know if we will receive one one or another
 639     if (sg_cfg_get_int("smpi/async_small_thres")>0){
 640         mailbox = smpi_process_mailbox_small();
 641         XBT_DEBUG("trying to probe the perm recv mailbox");
 642         request->action = simcall_comm_iprobe(mailbox, request->src, request->tag, &match_recv, (void*)request);
 643     }
 644     if (request->action==NULL){
 645         mailbox = smpi_process_mailbox();
 646         XBT_DEBUG("trying to probe the other mailbox");
 647         request->action = simcall_comm_iprobe(mailbox, request->src, request->tag, &match_recv, (void*)request);
 648     }
 649
 650   if(request->action){
 651     MPI_Request req = (MPI_Request)SIMIX_comm_get_src_data(request->action);
 652     *flag = 1;
 653     if(status != MPI_STATUS_IGNORE) {
 654       status->MPI_SOURCE = req->src;
 655       status->MPI_TAG = req->tag;
 656       status->MPI_ERROR = MPI_SUCCESS;
 657       status->count = req->real_size;
 658     }
 659   }
 660   else *flag = 0;
 661   smpi_mpi_request_free(&request);
 662
 663   return;
 664 }
 665
 666 void smpi_mpi_wait(MPI_Request * request, MPI_Status * status)
 667 {
 668   print_request("Waiting", *request);
 669   if ((*request)->action != NULL) { // this is not a detached send
 670     simcall_comm_wait((*request)->action, -1.0);
 671   }
 672   finish_wait(request, status);
 673
 674   // FIXME for a detached send, finish_wait is not called:
 675 }
 676
 677 int smpi_mpi_waitany(int count, MPI_Request requests[],
 678                      MPI_Status * status)
 679 {
 680   xbt_dynar_t comms;
 681   int i, size, index;
 682   int *map;
 683
 684   index = MPI_UNDEFINED;
 685   if(count > 0) {
 686     // Wait for a request to complete
 687     comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
 688     map = xbt_new(int, count);
 689     size = 0;
 690     XBT_DEBUG("Wait for one of %d", count);
 691     for(i = 0; i < count; i++) {
 692       if(requests[i] != MPI_REQUEST_NULL) {
 693         if (requests[i]->action != NULL) {
 694           XBT_DEBUG("Waiting any %p ", requests[i]);
 695           xbt_dynar_push(comms, &requests[i]->action);
 696           map[size] = i;
 697           size++;
 698         }else{
 699          //This is a finished detached request, let's return this one
 700          size=0;//so we free the dynar but don't do the waitany call
 701          index=i;
 702          finish_wait(&requests[i], status);//cleanup if refcount = 0
 703          requests[i]=MPI_REQUEST_NULL;//set to null
 704          break;
 705          }
 706       }
 707     }
 708     if(size > 0) {
 709       i = simcall_comm_waitany(comms);
 710
 711       // not MPI_UNDEFINED, as this is a simix return code
 712       if (i != -1) {
 713         index = map[i];
 714         finish_wait(&requests[index], status);
 715       }
 716     }
 717     xbt_free(map);
 718     xbt_dynar_free(&comms);
 719   }
 720
 721   if (index==MPI_UNDEFINED)
 722     smpi_empty_status(status);
 723
 724   return index;
 725 }
 726
 727 int smpi_mpi_waitall(int count, MPI_Request requests[],
 728                       MPI_Status status[])
 729 {
 730   int  index, c;
 731   MPI_Status stat;
 732   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 733   int retvalue=MPI_SUCCESS;
 734   //tag invalid requests in the set
 735   for(c = 0; c < count; c++) {
 736     if(requests[c]==MPI_REQUEST_NULL || requests[c]->dst == MPI_PROC_NULL ){
 737       if(status != MPI_STATUSES_IGNORE)
 738         smpi_empty_status(&status[c]);
 739     }else if(requests[c]->src == MPI_PROC_NULL ){
 740       if(status != MPI_STATUSES_IGNORE) {
 741         smpi_empty_status(&status[c]);
 742         status[c].MPI_SOURCE=MPI_PROC_NULL;
 743       }
 744     }
 745   }
 746   for(c = 0; c < count; c++) {
 747       if(MC_is_active()) {
 748         smpi_mpi_wait(&requests[c], pstat);
 749         index = c;
 750       } else {
 751         index = smpi_mpi_waitany(count, requests, pstat);
 752         if(index == MPI_UNDEFINED) {
 753           break;
 754        }
 755       if(status != MPI_STATUSES_IGNORE) {
 756         memcpy(&status[index], pstat, sizeof(*pstat));
 757         if(status[index].MPI_ERROR==MPI_ERR_TRUNCATE)retvalue=MPI_ERR_IN_STATUS;
 758
 759       }
 760     }
 761   }
 762
 763   return retvalue;
 764 }
 765
 766 int smpi_mpi_waitsome(int incount, MPI_Request requests[], int *indices,
 767                       MPI_Status status[])
 768 {
 769   int i, count, index;
 770   MPI_Status stat;
 771   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 772
 773   count = 0;
 774   for(i = 0; i < incount; i++)
 775   {
 776     index=smpi_mpi_waitany(incount, requests, pstat);
 777     if(index!=MPI_UNDEFINED){
 778       indices[count] = index;
 779       count++;
 780       if(status != MPI_STATUSES_IGNORE) {
 781         memcpy(&status[index], pstat, sizeof(*pstat));
 782       }
 783     }else{
 784       return MPI_UNDEFINED;
 785     }
 786   }
 787   return count;
 788 }
 789
 790 int smpi_mpi_testsome(int incount, MPI_Request requests[], int *indices,
 791                       MPI_Status status[])
 792 {
 793   int i, count, count_dead;
 794   MPI_Status stat;
 795   MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
 796
 797   count = 0;
 798   count_dead = 0;
 799   for(i = 0; i < incount; i++) {
 800     if((requests[i] != MPI_REQUEST_NULL)) {
 801       if(smpi_mpi_test(&requests[i], pstat)) {
 802          indices[count] = i;
 803          count++;
 804          if(status != MPI_STATUSES_IGNORE) {
 805             memcpy(&status[i], pstat, sizeof(*pstat));
 806          }
 807       }
 808     }else{
 809       count_dead++;
 810     }
 811   }
 812   if(count_dead==incount)return MPI_UNDEFINED;
 813   else return count;
 814 }
 815
 816 void smpi_mpi_bcast(void *buf, int count, MPI_Datatype datatype, int root,
 817                     MPI_Comm comm)
 818 {
 819   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
 820   nary_tree_bcast(buf, count, datatype, root, comm, 4);
 821 }
 822
 823 void smpi_mpi_barrier(MPI_Comm comm)
 824 {
 825   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
 826   nary_tree_barrier(comm, 4);
 827 }
 828
 829 void smpi_mpi_gather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 830                      void *recvbuf, int recvcount, MPI_Datatype recvtype,
 831                      int root, MPI_Comm comm)
 832 {
 833   int system_tag = 666;
 834   int rank, size, src, index;
 835   MPI_Aint lb = 0, recvext = 0;
 836   MPI_Request *requests;
 837
 838   rank = smpi_comm_rank(comm);
 839   size = smpi_comm_size(comm);
 840   if(rank != root) {
 841     // Send buffer to root
 842     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
 843   } else {
 844     // FIXME: check for errors
 845     smpi_datatype_extent(recvtype, &lb, &recvext);
 846     // Local copy from root
 847     smpi_datatype_copy(sendbuf, sendcount, sendtype,
 848                        (char *)recvbuf + root * recvcount * recvext, recvcount, recvtype);
 849     // Receive buffers from senders
 850     requests = xbt_new(MPI_Request, size - 1);
 851     index = 0;
 852     for(src = 0; src < size; src++) {
 853       if(src != root) {
 854         requests[index] = smpi_irecv_init((char *)recvbuf + src * recvcount * recvext,
 855                                           recvcount, recvtype,
 856                                           src, system_tag, comm);
 857         index++;
 858       }
 859     }
 860     // Wait for completion of irecv's.
 861     smpi_mpi_startall(size - 1, requests);
 862     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 863     xbt_free(requests);
 864   }
 865 }
 866
 867 void smpi_mpi_gatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 868                       void *recvbuf, int *recvcounts, int *displs,
 869                       MPI_Datatype recvtype, int root, MPI_Comm comm)
 870 {
 871   int system_tag = 666;
 872   int rank, size, src, index;
 873   MPI_Aint lb = 0, recvext = 0;
 874   MPI_Request *requests;
 875
 876   rank = smpi_comm_rank(comm);
 877   size = smpi_comm_size(comm);
 878   if(rank != root) {
 879     // Send buffer to root
 880     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
 881   } else {
 882     // FIXME: check for errors
 883     smpi_datatype_extent(recvtype, &lb, &recvext);
 884     // Local copy from root
 885     smpi_datatype_copy(sendbuf, sendcount, sendtype,
 886                        (char *)recvbuf + displs[root] * recvext,
 887                        recvcounts[root], recvtype);
 888     // Receive buffers from senders
 889     requests = xbt_new(MPI_Request, size - 1);
 890     index = 0;
 891     for(src = 0; src < size; src++) {
 892       if(src != root) {
 893         requests[index] =
 894           smpi_irecv_init((char *)recvbuf + displs[src] * recvext,
 895                           recvcounts[src], recvtype, src, system_tag, comm);
 896         index++;
 897       }
 898     }
 899     // Wait for completion of irecv's.
 900     smpi_mpi_startall(size - 1, requests);
 901     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
 902     xbt_free(requests);
 903   }
 904 }
 905
 906 void smpi_mpi_allgather(void *sendbuf, int sendcount,
 907                         MPI_Datatype sendtype, void *recvbuf,
 908                         int recvcount, MPI_Datatype recvtype,
 909                         MPI_Comm comm)
 910 {
 911   int system_tag = 666;
 912   int rank, size, other, index;
 913   MPI_Aint lb = 0, recvext = 0;
 914   MPI_Request *requests;
 915
 916   rank = smpi_comm_rank(comm);
 917   size = smpi_comm_size(comm);
 918   // FIXME: check for errors
 919   smpi_datatype_extent(recvtype, &lb, &recvext);
 920   // Local copy from self
 921   smpi_datatype_copy(sendbuf, sendcount, sendtype,
 922                      (char *)recvbuf + rank * recvcount * recvext, recvcount,
 923                      recvtype);
 924   // Send/Recv buffers to/from others;
 925   requests = xbt_new(MPI_Request, 2 * (size - 1));
 926   index = 0;
 927   for(other = 0; other < size; other++) {
 928     if(other != rank) {
 929       requests[index] =
 930         smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag,
 931                         comm);
 932       index++;
 933       requests[index] = smpi_irecv_init((char *)recvbuf + other * recvcount * recvext,
 934                                         recvcount, recvtype, other,
 935                                         system_tag, comm);
 936       index++;
 937     }
 938   }
 939   // Wait for completion of all comms.
 940   smpi_mpi_startall(2 * (size - 1), requests);
 941   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
 942   xbt_free(requests);
 943 }
 944
 945 void smpi_mpi_allgatherv(void *sendbuf, int sendcount,
 946                          MPI_Datatype sendtype, void *recvbuf,
 947                          int *recvcounts, int *displs,
 948                          MPI_Datatype recvtype, MPI_Comm comm)
 949 {
 950   int system_tag = 666;
 951   int rank, size, other, index;
 952   MPI_Aint lb = 0, recvext = 0;
 953   MPI_Request *requests;
 954
 955   rank = smpi_comm_rank(comm);
 956   size = smpi_comm_size(comm);
 957   // FIXME: check for errors
 958   smpi_datatype_extent(recvtype, &lb, &recvext);
 959   // Local copy from self
 960   smpi_datatype_copy(sendbuf, sendcount, sendtype,
 961                      (char *)recvbuf + displs[rank] * recvext,
 962                      recvcounts[rank], recvtype);
 963   // Send buffers to others;
 964   requests = xbt_new(MPI_Request, 2 * (size - 1));
 965   index = 0;
 966   for(other = 0; other < size; other++) {
 967     if(other != rank) {
 968       requests[index] =
 969         smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag,
 970                         comm);
 971       index++;
 972       requests[index] =
 973         smpi_irecv_init((char *)recvbuf + displs[other] * recvext, recvcounts[other],
 974                         recvtype, other, system_tag, comm);
 975       index++;
 976     }
 977   }
 978   // Wait for completion of all comms.
 979   smpi_mpi_startall(2 * (size - 1), requests);
 980   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
 981   xbt_free(requests);
 982 }
 983
 984 void smpi_mpi_scatter(void *sendbuf, int sendcount, MPI_Datatype sendtype,
 985                       void *recvbuf, int recvcount, MPI_Datatype recvtype,
 986                       int root, MPI_Comm comm)
 987 {
 988   int system_tag = 666;
 989   int rank, size, dst, index;
 990   MPI_Aint lb = 0, sendext = 0;
 991   MPI_Request *requests;
 992
 993   rank = smpi_comm_rank(comm);
 994   size = smpi_comm_size(comm);
 995   if(rank != root) {
 996     // Recv buffer from root
 997     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm,
 998                   MPI_STATUS_IGNORE);
 999   } else {
1000     // FIXME: check for errors
1001     smpi_datatype_extent(sendtype, &lb, &sendext);
1002     // Local copy from root
1003     smpi_datatype_copy((char *)sendbuf + root * sendcount * sendext,
1004                        sendcount, sendtype, recvbuf, recvcount, recvtype);
1005     // Send buffers to receivers
1006     requests = xbt_new(MPI_Request, size - 1);
1007     index = 0;
1008     for(dst = 0; dst < size; dst++) {
1009       if(dst != root) {
1010         requests[index] = smpi_isend_init((char *)sendbuf + dst * sendcount * sendext,
1011                                           sendcount, sendtype, dst,
1012                                           system_tag, comm);
1013         index++;
1014       }
1015     }
1016     // Wait for completion of isend's.
1017     smpi_mpi_startall(size - 1, requests);
1018     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
1019     xbt_free(requests);
1020   }
1021 }
1022
1023 void smpi_mpi_scatterv(void *sendbuf, int *sendcounts, int *displs,
1024                        MPI_Datatype sendtype, void *recvbuf, int recvcount,
1025                        MPI_Datatype recvtype, int root, MPI_Comm comm)
1026 {
1027   int system_tag = 666;
1028   int rank, size, dst, index;
1029   MPI_Aint lb = 0, sendext = 0;
1030   MPI_Request *requests;
1031
1032   rank = smpi_comm_rank(comm);
1033   size = smpi_comm_size(comm);
1034   if(rank != root) {
1035     // Recv buffer from root
1036     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm,
1037                   MPI_STATUS_IGNORE);
1038   } else {
1039     // FIXME: check for errors
1040     smpi_datatype_extent(sendtype, &lb, &sendext);
1041     // Local copy from root
1042     smpi_datatype_copy((char *)sendbuf + displs[root] * sendext, sendcounts[root],
1043                        sendtype, recvbuf, recvcount, recvtype);
1044     // Send buffers to receivers
1045     requests = xbt_new(MPI_Request, size - 1);
1046     index = 0;
1047     for(dst = 0; dst < size; dst++) {
1048       if(dst != root) {
1049         requests[index] =
1050           smpi_isend_init((char *)sendbuf + displs[dst] * sendext, sendcounts[dst],
1051                           sendtype, dst, system_tag, comm);
1052         index++;
1053       }
1054     }
1055     // Wait for completion of isend's.
1056     smpi_mpi_startall(size - 1, requests);
1057     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
1058     xbt_free(requests);
1059   }
1060 }
1061
1062 void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count,
1063                      MPI_Datatype datatype, MPI_Op op, int root,
1064                      MPI_Comm comm)
1065 {
1066   int system_tag = 666;
1067   int rank, size, src, index;
1068   MPI_Aint lb = 0, dataext = 0;
1069   MPI_Request *requests;
1070   void **tmpbufs;
1071
1072   rank = smpi_comm_rank(comm);
1073   size = smpi_comm_size(comm);
1074   if(rank != root) {
1075     // Send buffer to root
1076     smpi_mpi_send(sendbuf, count, datatype, root, system_tag, comm);
1077   } else {
1078     // FIXME: check for errors
1079     smpi_datatype_extent(datatype, &lb, &dataext);
1080     // Local copy from root
1081     if (sendbuf && recvbuf)
1082       smpi_datatype_copy(sendbuf, count, datatype, recvbuf, count, datatype);
1083     // Receive buffers from senders
1084     //TODO: make a MPI_barrier here ?
1085     requests = xbt_new(MPI_Request, size - 1);
1086     tmpbufs = xbt_new(void *, size - 1);
1087     index = 0;
1088     for(src = 0; src < size; src++) {
1089       if(src != root) {
1090         // FIXME: possibly overkill we we have contiguous/noncontiguous data
1091         //  mapping...
1092         tmpbufs[index] = xbt_malloc(count * dataext);
1093         requests[index] =
1094           smpi_irecv_init(tmpbufs[index], count, datatype, src,
1095                           system_tag, comm);
1096         index++;
1097       }
1098     }
1099     // Wait for completion of irecv's.
1100     smpi_mpi_startall(size - 1, requests);
1101     for(src = 0; src < size - 1; src++) {
1102       index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
1103       XBT_DEBUG("finished waiting any request with index %d", index);
1104       if(index == MPI_UNDEFINED) {
1105         break;
1106       }
1107       if(op) /* op can be MPI_OP_NULL that does nothing */
1108         smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
1109     }
1110     for(index = 0; index < size - 1; index++) {
1111       xbt_free(tmpbufs[index]);
1112     }
1113     xbt_free(tmpbufs);
1114     xbt_free(requests);
1115   }
1116 }
1117
1118 void smpi_mpi_allreduce(void *sendbuf, void *recvbuf, int count,
1119                         MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
1120 {
1121   smpi_mpi_reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
1122   smpi_mpi_bcast(recvbuf, count, datatype, 0, comm);
1123 }
1124
1125 void smpi_mpi_scan(void *sendbuf, void *recvbuf, int count,
1126                    MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
1127 {
1128   int system_tag = 666;
1129   int rank, size, other, index;
1130   MPI_Aint lb = 0, dataext = 0;
1131   MPI_Request *requests;
1132   void **tmpbufs;
1133
1134   rank = smpi_comm_rank(comm);
1135   size = smpi_comm_size(comm);
1136
1137   // FIXME: check for errors
1138   smpi_datatype_extent(datatype, &lb, &dataext);
1139
1140   // Local copy from self
1141   smpi_datatype_copy(sendbuf, count, datatype, recvbuf, count, datatype);
1142
1143   // Send/Recv buffers to/from others;
1144   requests = xbt_new(MPI_Request, size - 1);
1145   tmpbufs = xbt_new(void *, rank);
1146   index = 0;
1147   for(other = 0; other < rank; other++) {
1148     // FIXME: possibly overkill we we have contiguous/noncontiguous data
1149     // mapping...
1150     tmpbufs[index] = xbt_malloc(count * dataext);
1151     requests[index] =
1152       smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag,
1153                       comm);
1154     index++;
1155   }
1156   for(other = rank + 1; other < size; other++) {
1157     requests[index] =
1158       smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm);
1159     index++;
1160   }
1161   // Wait for completion of all comms.
1162   smpi_mpi_startall(size - 1, requests);
1163   for(other = 0; other < size - 1; other++) {
1164     index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
1165     if(index == MPI_UNDEFINED) {
1166       break;
1167     }
1168     if(index < rank) {
1169       // #Request is below rank: it's a irecv
1170       smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
1171     }
1172   }
1173   for(index = 0; index < rank; index++) {
1174     xbt_free(tmpbufs[index]);
1175   }
1176   xbt_free(tmpbufs);
1177   xbt_free(requests);
1178 }