src/smpi/smpi_replay.c

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