src/smpi/smpi_replay.c

   1 /* Copyright (c) 2009, 2010, 2011, 2012. 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
   8 #include "private.h"
   9 #include <stdio.h>
  10 #include <xbt.h>
  11 #include <xbt/replay.h>
  12
  13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_replay,smpi,"Trace Replay with SMPI");
  14
  15 int communicator_size = 0;
  16 static int active_processes = 0;
  17 xbt_dynar_t *reqq = NULL;
  18
  19 MPI_Datatype MPI_DEFAULT_TYPE;
  20 MPI_Datatype MPI_CURRENT_TYPE;
  21
  22 static void log_timed_action (const char *const *action, double clock){
  23   if (XBT_LOG_ISENABLED(smpi_replay, xbt_log_priority_verbose)){
  24     char *name = xbt_str_join_array(action, " ");
  25     XBT_VERB("%s %f", name, smpi_process_simulated_elapsed()-clock);
  26     free(name);
  27   }
  28 }
  29
  30 typedef struct {
  31   xbt_dynar_t isends; /* of MPI_Request */
  32   xbt_dynar_t irecvs; /* of MPI_Request */
  33 } s_smpi_replay_globals_t, *smpi_replay_globals_t;
  34
  35
  36 /* Helper function */
  37 static double parse_double(const char *string)
  38 {
  39   double value;
  40   char *endptr;
  41   value = strtod(string, &endptr);
  42   if (*endptr != '\0')
  43     THROWF(unknown_error, 0, "%s is not a double", string);
  44   return value;
  45 }
  46
  47 static MPI_Datatype decode_datatype(const char *const action)
  48 {
  49 // Declared datatypes,
  50
  51   switch(atoi(action))
  52   {
  53     case 0:
  54       MPI_CURRENT_TYPE=MPI_DOUBLE;
  55       break;
  56     case 1:
  57       MPI_CURRENT_TYPE=MPI_INT;
  58       break;
  59     case 2:
  60       MPI_CURRENT_TYPE=MPI_CHAR;
  61       break;
  62     case 3:
  63       MPI_CURRENT_TYPE=MPI_SHORT;
  64       break;
  65     case 4:
  66       MPI_CURRENT_TYPE=MPI_LONG;
  67       break;
  68     case 5:
  69       MPI_CURRENT_TYPE=MPI_FLOAT;
  70       break;
  71     case 6:
  72       MPI_CURRENT_TYPE=MPI_BYTE;
  73       break;
  74     default:
  75       MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
  76
  77   }
  78    return MPI_CURRENT_TYPE;
  79 }
  80
  81 static void action_init(const char *const *action)
  82 {
  83   int i;
  84   XBT_DEBUG("Initialize the counters");
  85   smpi_replay_globals_t globals =  xbt_new(s_smpi_replay_globals_t, 1);
  86   globals->isends = xbt_dynar_new(sizeof(MPI_Request),NULL);
  87   globals->irecvs = xbt_dynar_new(sizeof(MPI_Request),NULL);
  88
  89   if(action[2]) MPI_DEFAULT_TYPE= MPI_DOUBLE; // default MPE dataype
  90   else MPI_DEFAULT_TYPE= MPI_BYTE; // default TAU datatype
  91
  92   smpi_process_set_user_data((void*) globals);
  93
  94   /* start a simulated timer */
  95   smpi_process_simulated_start();
  96   /*initialize the number of active processes */
  97   active_processes = smpi_process_count();
  98
  99   if (!reqq) {
 100     reqq=xbt_new0(xbt_dynar_t,active_processes);
 101
 102     for(i=0;i<active_processes;i++){
 103       reqq[i]=xbt_dynar_new(sizeof(MPI_Request),NULL);
 104     }
 105   }
 106 }
 107
 108 static void action_finalize(const char *const *action)
 109 {
 110   smpi_replay_globals_t globals =
 111       (smpi_replay_globals_t) smpi_process_get_user_data();
 112   if (globals){
 113     XBT_DEBUG("There are %lu isends and %lu irecvs in the dynars",
 114          xbt_dynar_length(globals->isends),xbt_dynar_length(globals->irecvs));
 115     xbt_dynar_free_container(&(globals->isends));
 116     xbt_dynar_free_container(&(globals->irecvs));
 117   }
 118   free(globals);
 119 }
 120
 121 static void action_comm_size(const char *const *action)
 122 {
 123   double clock = smpi_process_simulated_elapsed();
 124
 125   communicator_size = parse_double(action[2]);
 126   log_timed_action (action, clock);
 127 }
 128
 129 static void action_comm_split(const char *const *action)
 130 {
 131   double clock = smpi_process_simulated_elapsed();
 132
 133   log_timed_action (action, clock);
 134 }
 135
 136 static void action_comm_dup(const char *const *action)
 137 {
 138   double clock = smpi_process_simulated_elapsed();
 139
 140   log_timed_action (action, clock);
 141 }
 142
 143 static void action_compute(const char *const *action)
 144 {
 145   double clock = smpi_process_simulated_elapsed();
 146   smpi_execute_flops(parse_double(action[2]));
 147
 148   log_timed_action (action, clock);
 149 }
 150
 151 static void action_send(const char *const *action)
 152 {
 153   int to = atoi(action[2]);
 154   double size=parse_double(action[3]);
 155   double clock = smpi_process_simulated_elapsed();
 156
 157   if(action[4]) {
 158     MPI_CURRENT_TYPE=decode_datatype(action[4]);
 159   } else {
 160     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 161   }
 162
 163 #ifdef HAVE_TRACING
 164   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 165   TRACE_smpi_computing_out(rank);
 166   int dst_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), to);
 167   TRACE_smpi_ptp_in(rank, rank, dst_traced, __FUNCTION__);
 168   TRACE_smpi_send(rank, rank, dst_traced);
 169 #endif
 170
 171   smpi_mpi_send(NULL, size, MPI_CURRENT_TYPE, to , 0, MPI_COMM_WORLD);
 172
 173   log_timed_action (action, clock);
 174
 175   #ifdef HAVE_TRACING
 176   TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
 177   TRACE_smpi_computing_in(rank);
 178 #endif
 179
 180 }
 181
 182 static void action_Isend(const char *const *action)
 183 {
 184   int to = atoi(action[2]);
 185   double size=parse_double(action[3]);
 186   double clock = smpi_process_simulated_elapsed();
 187   MPI_Request request;
 188
 189   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 190   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 191
 192   smpi_replay_globals_t globals =
 193      (smpi_replay_globals_t) smpi_process_get_user_data();
 194 #ifdef HAVE_TRACING
 195   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 196   TRACE_smpi_computing_out(rank);
 197   int dst_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), to);
 198   TRACE_smpi_ptp_in(rank, rank, dst_traced, __FUNCTION__);
 199   TRACE_smpi_send(rank, rank, dst_traced);
 200 #endif
 201
 202   request = smpi_mpi_isend(NULL, size, MPI_CURRENT_TYPE, to, 0,MPI_COMM_WORLD);
 203
 204 #ifdef HAVE_TRACING
 205   TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
 206   request->send = 1;
 207   TRACE_smpi_computing_in(rank);
 208 #endif
 209
 210   xbt_dynar_push(globals->isends,&request);
 211   xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
 212
 213   log_timed_action (action, clock);
 214 }
 215
 216 static void action_recv(const char *const *action) {
 217   int from = atoi(action[2]);
 218   double size=parse_double(action[3]);
 219   double clock = smpi_process_simulated_elapsed();
 220   MPI_Status status;
 221
 222   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 223   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 224
 225 #ifdef HAVE_TRACING
 226   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 227   int src_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), from);
 228   TRACE_smpi_computing_out(rank);
 229
 230   TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__);
 231 #endif
 232
 233   smpi_mpi_recv(NULL, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD, &status);
 234
 235 #ifdef HAVE_TRACING
 236   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
 237   TRACE_smpi_recv(rank, src_traced, rank);
 238   TRACE_smpi_computing_in(rank);
 239 #endif
 240
 241   log_timed_action (action, clock);
 242 }
 243
 244 static void action_Irecv(const char *const *action)
 245 {
 246   int from = atoi(action[2]);
 247   double size=parse_double(action[3]);
 248   double clock = smpi_process_simulated_elapsed();
 249   MPI_Request request;
 250
 251   smpi_replay_globals_t globals =
 252      (smpi_replay_globals_t) smpi_process_get_user_data();
 253
 254   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 255   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 256
 257 #ifdef HAVE_TRACING
 258   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 259   int src_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), from);
 260   TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__);
 261 #endif
 262
 263   request = smpi_mpi_irecv(NULL, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD);
 264
 265 #ifdef HAVE_TRACING
 266   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
 267   request->recv = 1;
 268 #endif
 269   xbt_dynar_push(globals->irecvs,&request);
 270   xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
 271
 272   log_timed_action (action, clock);
 273 }
 274
 275 static void action_wait(const char *const *action){
 276   double clock = smpi_process_simulated_elapsed();
 277   MPI_Request request;
 278   MPI_Status status;
 279   smpi_replay_globals_t globals =
 280       (smpi_replay_globals_t) smpi_process_get_user_data();
 281
 282   xbt_assert(xbt_dynar_length(globals->irecvs),
 283       "action wait not preceded by any irecv: %s",
 284       xbt_str_join_array(action," "));
 285   request = xbt_dynar_pop_as(globals->irecvs,MPI_Request);
 286 #ifdef HAVE_TRACING
 287   int rank = request && request->comm != MPI_COMM_NULL
 288       ? smpi_comm_rank(request->comm)
 289       : -1;
 290   TRACE_smpi_computing_out(rank);
 291
 292   MPI_Group group = smpi_comm_group(request->comm);
 293   int src_traced = smpi_group_rank(group, request->src);
 294   int dst_traced = smpi_group_rank(group, request->dst);
 295   int is_wait_for_receive = request->recv;
 296   TRACE_smpi_ptp_in(rank, src_traced, dst_traced, __FUNCTION__);
 297 #endif
 298   smpi_mpi_wait(&request, &status);
 299 #ifdef HAVE_TRACING
 300   TRACE_smpi_ptp_out(rank, src_traced, dst_traced, __FUNCTION__);
 301   if (is_wait_for_receive) {
 302     TRACE_smpi_recv(rank, src_traced, dst_traced);
 303   }
 304   TRACE_smpi_computing_in(rank);
 305 #endif
 306
 307   log_timed_action (action, clock);
 308 }
 309
 310 static void action_waitall(const char *const *action){
 311   double clock = smpi_process_simulated_elapsed();
 312   int count_requests=0;
 313   unsigned int i=0;
 314
 315   count_requests=xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 316
 317   if (count_requests>0) {
 318     MPI_Request requests[count_requests];
 319     MPI_Status status[count_requests];
 320
 321     /*  The reqq is an array of dynars. Its index corresponds to the rank.
 322      Thus each rank saves its own requests to the array request. */
 323     xbt_dynar_foreach(reqq[smpi_comm_rank(MPI_COMM_WORLD)],i,requests[i]);
 324
 325   #ifdef HAVE_TRACING
 326    //save information from requests
 327
 328    xbt_dynar_t srcs = xbt_dynar_new(sizeof(int), NULL);
 329    xbt_dynar_t dsts = xbt_dynar_new(sizeof(int), NULL);
 330    xbt_dynar_t recvs = xbt_dynar_new(sizeof(int), NULL);
 331    for (i = 0; i < count_requests; i++) {
 332     if(requests[i]){
 333       int *asrc = xbt_new(int, 1);
 334       int *adst = xbt_new(int, 1);
 335       int *arecv = xbt_new(int, 1);
 336       *asrc = requests[i]->src;
 337       *adst = requests[i]->dst;
 338       *arecv = requests[i]->recv;
 339       xbt_dynar_insert_at(srcs, i, asrc);
 340       xbt_dynar_insert_at(dsts, i, adst);
 341       xbt_dynar_insert_at(recvs, i, arecv);
 342       xbt_free(asrc);
 343       xbt_free(adst);
 344       xbt_free(arecv);
 345     }else {
 346       int *t = xbt_new(int, 1);
 347       xbt_dynar_insert_at(srcs, i, t);
 348       xbt_dynar_insert_at(dsts, i, t);
 349       xbt_dynar_insert_at(recvs, i, t);
 350       xbt_free(t);
 351     }
 352    }
 353    int rank_traced = smpi_process_index();
 354    TRACE_smpi_computing_out(rank_traced);
 355
 356    TRACE_smpi_ptp_in(rank_traced, -1, -1, __FUNCTION__);
 357   #endif
 358
 359     smpi_mpi_waitall(count_requests, requests, status);
 360
 361   #ifdef HAVE_TRACING
 362    for (i = 0; i < count_requests; i++) {
 363     int src_traced, dst_traced, is_wait_for_receive;
 364     xbt_dynar_get_cpy(srcs, i, &src_traced);
 365     xbt_dynar_get_cpy(dsts, i, &dst_traced);
 366     xbt_dynar_get_cpy(recvs, i, &is_wait_for_receive);
 367     if (is_wait_for_receive) {
 368       TRACE_smpi_recv(rank_traced, src_traced, dst_traced);
 369     }
 370    }
 371    TRACE_smpi_ptp_out(rank_traced, -1, -1, __FUNCTION__);
 372    //clean-up of dynars
 373    xbt_dynar_free(&srcs);
 374    xbt_dynar_free(&dsts);
 375    xbt_dynar_free(&recvs);
 376    TRACE_smpi_computing_in(rank_traced);
 377   #endif
 378
 379    xbt_dynar_reset(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 380   }
 381   log_timed_action (action, clock);
 382 }
 383
 384 static void action_barrier(const char *const *action){
 385   double clock = smpi_process_simulated_elapsed();
 386 #ifdef HAVE_TRACING
 387   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 388   TRACE_smpi_computing_out(rank);
 389   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 390 #endif
 391   smpi_mpi_barrier(MPI_COMM_WORLD);
 392 #ifdef HAVE_TRACING
 393   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 394   TRACE_smpi_computing_in(rank);
 395 #endif
 396
 397   log_timed_action (action, clock);
 398 }
 399
 400
 401 static void action_bcast(const char *const *action)
 402 {
 403   double size = parse_double(action[2]);
 404   double clock = smpi_process_simulated_elapsed();
 405   int root=0;
 406   /*
 407    * Initialize MPI_CURRENT_TYPE in order to decrease
 408    * the number of the checks
 409    * */
 410   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 411
 412   if(action[3]) {
 413     root= atoi(action[3]);
 414     if(action[4]) {
 415       MPI_CURRENT_TYPE=decode_datatype(action[4]);
 416     }
 417   }
 418
 419 #ifdef HAVE_TRACING
 420   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 421   TRACE_smpi_computing_out(rank);
 422   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), 0);
 423   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__);
 424 #endif
 425
 426   smpi_mpi_bcast(NULL, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
 427 #ifdef HAVE_TRACING
 428   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 429   TRACE_smpi_computing_in(rank);
 430 #endif
 431
 432   log_timed_action (action, clock);
 433 }
 434
 435 static void action_reduce(const char *const *action)
 436 {
 437   double comm_size = parse_double(action[2]);
 438   double comp_size = parse_double(action[3]);
 439   double clock = smpi_process_simulated_elapsed();
 440   int root=0;
 441   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 442
 443   if(action[4]) {
 444       root= atoi(action[4]);
 445       if(action[5]) {
 446         MPI_CURRENT_TYPE=decode_datatype(action[5]);
 447       }
 448   }
 449
 450 #ifdef HAVE_TRACING
 451   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 452   TRACE_smpi_computing_out(rank);
 453   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), 0);
 454   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__);
 455 #endif
 456    mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
 457    smpi_execute_flops(comp_size);
 458 #ifdef HAVE_TRACING
 459   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 460   TRACE_smpi_computing_in(rank);
 461 #endif
 462
 463   log_timed_action (action, clock);
 464 }
 465
 466 static void action_allReduce(const char *const *action) {
 467   double comm_size = parse_double(action[2]);
 468   double comp_size = parse_double(action[3]);
 469
 470   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 471   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 472
 473   double clock = smpi_process_simulated_elapsed();
 474 #ifdef HAVE_TRACING
 475   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 476   TRACE_smpi_computing_out(rank);
 477   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 478 #endif
 479    mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
 480   smpi_execute_flops(comp_size);
 481    mpi_coll_bcast_fun(NULL, comm_size, MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
 482 #ifdef HAVE_TRACING
 483   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 484   TRACE_smpi_computing_in(rank);
 485 #endif
 486
 487   log_timed_action (action, clock);
 488 }
 489
 490 static void action_allToAll(const char *const *action) {
 491   double clock = smpi_process_simulated_elapsed();
 492   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 493   int send_size = parse_double(action[2]);
 494   int recv_size = parse_double(action[3]);
 495   MPI_Datatype MPI_CURRENT_TYPE2;
 496
 497   if(action[4]) {
 498     MPI_CURRENT_TYPE=decode_datatype(action[4]);
 499     MPI_CURRENT_TYPE2=decode_datatype(action[5]);
 500   }
 501   else {
 502     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 503     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 504   }
 505   void *send = calloc(send_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 506   void *recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 507
 508 #ifdef HAVE_TRACING
 509   int rank = smpi_process_index();
 510   TRACE_smpi_computing_out(rank);
 511   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 512 #endif
 513
 514   mpi_coll_alltoall_fun(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
 515
 516 #ifdef HAVE_TRACING
 517   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 518   TRACE_smpi_computing_in(rank);
 519 #endif
 520
 521   log_timed_action (action, clock);
 522   xbt_free(send);
 523   xbt_free(recv);
 524 }
 525
 526
 527 static void action_gather(const char *const *action) {
 528
 529
 530   /*
 531  The structure of the gather action for the rank 0 (total 4 processes)
 532  is the following:
 533  0 gather 68 68 0 0 0
 534
 535   where:
 536   1) 68 is the sendcounts
 537   2) 68 is the recvcounts
 538   3) 0 is the root node
 539   4) 0 is the send datatype id, see decode_datatype()
 540   5) 0 is the recv datatype id, see decode_datatype()
 541
 542   */
 543   double clock = smpi_process_simulated_elapsed();
 544   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 545   int send_size = parse_double(action[2]);
 546   int recv_size = parse_double(action[3]);
 547   MPI_Datatype MPI_CURRENT_TYPE2;
 548   if(action[5]) {
 549     MPI_CURRENT_TYPE=decode_datatype(action[5]);
 550     MPI_CURRENT_TYPE2=decode_datatype(action[6]);
 551   } else {
 552     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 553     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 554   }
 555   void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 556   void *recv = calloc(recv_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 557
 558   int root=atoi(action[4]);
 559   int rank = smpi_process_index();
 560
 561   if(rank==root) recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 562
 563
 564 #ifdef HAVE_TRACING
 565   int rank = smpi_process_index();
 566   TRACE_smpi_computing_out(rank);
 567   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 568 #endif
 569 smpi_mpi_gather(send, send_size, MPI_CURRENT_TYPE,
 570                 recv, recv_size, MPI_CURRENT_TYPE2,
 571                 root, MPI_COMM_WORLD);
 572
 573 #ifdef HAVE_TRACING
 574   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 575   TRACE_smpi_computing_in(rank);
 576 #endif
 577
 578   log_timed_action (action, clock);
 579   xbt_free(send);
 580   xbt_free(recv);
 581 }
 582
 583
 584 static void action_reducescatter(const char *const *action) {
 585
 586     /*
 587  The structure of the reducescatter action for the rank 0 (total 4 processes)
 588  is the following:
 589 0 reduceScatter 275427 275427 275427 204020 11346849 0
 590
 591   where:
 592   1) The first four values after the name of the action declare the recvcounts array
 593   2) The value 11346849 is the amount of instructions
 594   3) The last value corresponds to the datatype, see decode_datatype().
 595
 596   We analyze a MPI_Reduce_scatter call to one MPI_Reduce and one MPI_Scatterv.
 597
 598    */
 599
 600   double clock = smpi_process_simulated_elapsed();
 601   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 602   int comp_size = parse_double(action[2+comm_size]);
 603   int *recvcounts = xbt_new0(int, comm_size);
 604   int *disps = xbt_new0(int, comm_size);
 605   int i=0,recv_sum=0;
 606   int root=0;
 607   int rank = smpi_process_index();
 608
 609   if(action[3+comm_size]) MPI_CURRENT_TYPE=decode_datatype(action[3+comm_size]);
 610   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 611
 612   for(i=0;i<comm_size;i++) {
 613     recvcounts[i] = atoi(action[i+2]);
 614     recv_sum=recv_sum+recvcounts[i];
 615     disps[i] = 0;
 616   }
 617
 618 #ifdef HAVE_TRACING
 619   int rank = smpi_process_index();
 620   TRACE_smpi_computing_out(rank);
 621   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 622 #endif
 623    mpi_coll_reduce_fun(NULL, NULL, recv_sum, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
 624    smpi_mpi_scatterv(NULL, recvcounts, disps, MPI_CURRENT_TYPE, NULL,
 625                       recvcounts[rank], MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
 626    smpi_execute_flops(comp_size);
 627
 628
 629 #ifdef HAVE_TRACING
 630   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 631   TRACE_smpi_computing_in(rank);
 632 #endif
 633
 634   log_timed_action (action, clock);
 635 }
 636
 637
 638 static void action_allgatherv(const char *const *action) {
 639
 640   /*
 641  The structure of the allgatherv action for the rank 0 (total 4 processes)
 642  is the following:
 643 0 allGatherV 275427 275427 275427 275427 204020 0 275427 550854 826281
 644
 645   where:
 646   1) 275427 is the sendcount
 647   2) The next four elements declare the recvcounts array
 648   3) The next four values declare the disps array
 649   4) No more values mean that the datatype for sent and receive buffer
 650   is the default one, see decode_datatype().
 651
 652    */
 653
 654   double clock = smpi_process_simulated_elapsed();
 655
 656   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 657   int i=0;
 658   int sendcount=atoi(action[2]);
 659   int *recvcounts = xbt_new0(int, comm_size);
 660   int *disps = xbt_new0(int, comm_size);
 661   int recv_sum=0;
 662   MPI_Datatype MPI_CURRENT_TYPE2;
 663
 664   if(action[3+2*comm_size]) {
 665     MPI_CURRENT_TYPE = decode_datatype(action[3+2*comm_size]);
 666     MPI_CURRENT_TYPE2 = decode_datatype(action[4+2*comm_size]);
 667   } else {
 668     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
 669     MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
 670   }
 671   void *sendbuf = calloc(sendcount, smpi_datatype_size(MPI_CURRENT_TYPE));
 672
 673   for(i=0;i<comm_size;i++) {
 674     recvcounts[i] = atoi(action[i+3]);
 675     recv_sum=recv_sum+recvcounts[i];
 676     disps[i] = atoi(action[i+3+comm_size]);
 677   }
 678   void *recvbuf = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));
 679
 680 #ifdef HAVE_TRACING
 681   int rank = MPI_COMM_WORLD != MPI_COMM_NULL ? smpi_process_index() : -1;
 682   TRACE_smpi_computing_out(rank);
 683   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 684 #endif
 685
 686 mpi_coll_allgatherv_fun(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcounts, disps, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
 687
 688 #ifdef HAVE_TRACING
 689   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 690   TRACE_smpi_computing_in(rank);
 691 #endif
 692
 693   log_timed_action (action, clock);
 694   xbt_free(sendbuf);
 695   xbt_free(recvbuf);
 696   xbt_free(recvcounts);
 697   xbt_free(disps);
 698 }
 699
 700
 701 static void action_allToAllv(const char *const *action) {
 702   /*
 703  The structure of the allToAllV action for the rank 0 (total 4 processes)
 704  is the following:
 705   0 allToAllV 100 1 7 10 12 5 10 20 45 100 1 70 10 5 1 5 77 90
 706
 707   where:
 708   1) 100 is the size of the send buffer *sizeof(int),
 709   2) 1 7 10 12 is the sendcounts array
 710   3) 5 10 20 45 is the sdispls array
 711   4) 100*sizeof(int) is the size of the receiver buffer
 712   5)  1 70 10 5 is the recvcounts array
 713   6) 1 5 77 90 is the rdispls array
 714
 715    */
 716
 717
 718   double clock = smpi_process_simulated_elapsed();
 719
 720   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 721   int send_buf_size=0,recv_buf_size=0,i=0;
 722   int *sendcounts = xbt_new0(int, comm_size);
 723   int *recvcounts = xbt_new0(int, comm_size);
 724   int *senddisps = xbt_new0(int, comm_size);
 725   int *recvdisps = xbt_new0(int, comm_size);
 726
 727   MPI_Datatype MPI_CURRENT_TYPE2;
 728
 729   send_buf_size=parse_double(action[2]);
 730   recv_buf_size=parse_double(action[3+2*comm_size]);
 731   if(action[4+4*comm_size]) {
 732     MPI_CURRENT_TYPE=decode_datatype(action[4+4*comm_size]);
 733     MPI_CURRENT_TYPE2=decode_datatype(action[5+4*comm_size]);
 734   }
 735   else {
 736       MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 737       MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 738   }
 739
 740   void *sendbuf = calloc(send_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 741   void *recvbuf = calloc(recv_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 742
 743   for(i=0;i<comm_size;i++) {
 744     sendcounts[i] = atoi(action[i+3]);
 745     senddisps[i] = atoi(action[i+3+comm_size]);
 746     recvcounts[i] = atoi(action[i+4+2*comm_size]);
 747     recvdisps[i] = atoi(action[i+4+3*comm_size]);
 748   }
 749
 750
 751 #ifdef HAVE_TRACING
 752   int rank = MPI_COMM_WORLD != MPI_COMM_NULL ? smpi_process_index() : -1;
 753   TRACE_smpi_computing_out(rank);
 754   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 755 #endif
 756     mpi_coll_alltoallv_fun(sendbuf, sendcounts, senddisps,      MPI_CURRENT_TYPE,
 757                                recvbuf, recvcounts, recvdisps, MPI_CURRENT_TYPE,
 758                                MPI_COMM_WORLD);
 759 #ifdef HAVE_TRACING
 760   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 761   TRACE_smpi_computing_in(rank);
 762 #endif
 763
 764   log_timed_action (action, clock);
 765   xbt_free(sendbuf);
 766   xbt_free(recvbuf);
 767   xbt_free(sendcounts);
 768   xbt_free(recvcounts);
 769   xbt_free(senddisps);
 770   xbt_free(recvdisps);
 771
 772
 773 }
 774
 775 void smpi_replay_init(int *argc, char***argv){
 776   PMPI_Init(argc, argv);
 777   if (!smpi_process_index()){
 778     _xbt_replay_action_init();
 779     xbt_replay_action_register("init",       action_init);
 780     xbt_replay_action_register("finalize",   action_finalize);
 781     xbt_replay_action_register("comm_size",  action_comm_size);
 782     xbt_replay_action_register("comm_split", action_comm_split);
 783     xbt_replay_action_register("comm_dup",   action_comm_dup);
 784     xbt_replay_action_register("send",       action_send);
 785     xbt_replay_action_register("Isend",      action_Isend);
 786     xbt_replay_action_register("recv",       action_recv);
 787     xbt_replay_action_register("Irecv",      action_Irecv);
 788     xbt_replay_action_register("wait",       action_wait);
 789     xbt_replay_action_register("waitAll",    action_waitall);
 790     xbt_replay_action_register("barrier",    action_barrier);
 791     xbt_replay_action_register("bcast",      action_bcast);
 792     xbt_replay_action_register("reduce",     action_reduce);
 793     xbt_replay_action_register("allReduce",  action_allReduce);
 794     xbt_replay_action_register("allToAll",   action_allToAll);
 795     xbt_replay_action_register("allToAllV",  action_allToAllv);
 796     xbt_replay_action_register("gather",  action_gather);
 797     xbt_replay_action_register("allGatherV",  action_allgatherv);
 798     xbt_replay_action_register("reduceScatter",  action_reducescatter);
 799     xbt_replay_action_register("compute",    action_compute);
 800   }
 801
 802   xbt_replay_action_runner(*argc, *argv);
 803 }
 804
 805 int smpi_replay_finalize(){
 806   double sim_time= 1.;
 807   /* One active process will stop. Decrease the counter*/
 808   active_processes--;
 809   XBT_DEBUG("There are %lu elements in reqq[*]",
 810             xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]));
 811   xbt_dynar_free(&reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 812   if(!active_processes){
 813     /* Last process alive speaking */
 814     /* end the simulated timer */
 815     sim_time = smpi_process_simulated_elapsed();
 816     XBT_INFO("Simulation time %g", sim_time);
 817     _xbt_replay_action_exit();
 818     xbt_free(reqq);
 819     reqq = NULL;
 820   }
 821   smpi_mpi_barrier(MPI_COMM_WORLD);
 822   return PMPI_Finalize();
 823 }