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;
  18
  19 static void log_timed_action (const char *const *action, double clock){
  20   if (XBT_LOG_ISENABLED(smpi_replay, xbt_log_priority_verbose)){
  21     char *name = xbt_str_join_array(action, " ");
  22     XBT_VERB("%s %f", name, smpi_process_simulated_elapsed()-clock);
  23     free(name);
  24   }
  25 }
  26
  27
  28 typedef struct {
  29   xbt_dynar_t isends; /* of MPI_Request */
  30   xbt_dynar_t irecvs; /* of MPI_Request */
  31 } s_smpi_replay_globals_t, *smpi_replay_globals_t;
  32
  33
  34 /* Helper function */
  35 static double parse_double(const char *string)
  36 {
  37   double value;
  38   char *endptr;
  39   value = strtod(string, &endptr);
  40   if (*endptr != '\0')
  41     THROWF(unknown_error, 0, "%s is not a double", string);
  42   return value;
  43 }
  44
  45 static void action_init(const char *const *action)
  46 {
  47   int i;
  48   XBT_DEBUG("Initialize the counters");
  49   smpi_replay_globals_t globals =  xbt_new(s_smpi_replay_globals_t, 1);
  50   globals->isends = xbt_dynar_new(sizeof(MPI_Request),NULL);
  51   globals->irecvs = xbt_dynar_new(sizeof(MPI_Request),NULL);
  52
  53
  54   smpi_process_set_user_data((void*) globals);
  55
  56   /* start a simulated timer */
  57   smpi_process_simulated_start();
  58   /*initialize the number of active processes */
  59   active_processes = smpi_process_count();
  60
  61   reqq=xbt_new0(xbt_dynar_t,active_processes);
  62
  63   for(i=0;i<active_processes;i++){
  64     reqq[i]=xbt_dynar_new(sizeof(MPI_Request),NULL);
  65   }
  66
  67
  68 }
  69
  70 static void action_finalize(const char *const *action)
  71 {
  72   smpi_replay_globals_t globals =
  73       (smpi_replay_globals_t) smpi_process_get_user_data();
  74
  75   if (globals){
  76     XBT_DEBUG("There are %lu isends and %lu irecvs in the dynars",
  77          xbt_dynar_length(globals->isends),xbt_dynar_length(globals->irecvs));
  78     xbt_dynar_free_container(&(globals->isends));
  79     xbt_dynar_free_container(&(globals->irecvs));
  80   }
  81   free(globals);
  82 }
  83
  84 static void action_comm_size(const char *const *action)
  85 {
  86   double clock = smpi_process_simulated_elapsed();
  87
  88   communicator_size = parse_double(action[2]);
  89   log_timed_action (action, clock);
  90 }
  91
  92 static void action_comm_split(const char *const *action)
  93 {
  94   double clock = smpi_process_simulated_elapsed();
  95
  96   log_timed_action (action, clock);
  97 }
  98
  99 static void action_comm_dup(const char *const *action)
 100 {
 101   double clock = smpi_process_simulated_elapsed();
 102
 103   log_timed_action (action, clock);
 104 }
 105
 106 static void action_compute(const char *const *action)
 107 {
 108   double clock = smpi_process_simulated_elapsed();
 109   smpi_execute_flops(parse_double(action[2]));
 110
 111   log_timed_action (action, clock);
 112 }
 113
 114 static void action_send(const char *const *action)
 115 {
 116   int to = atoi(action[2]);
 117   double size=parse_double(action[3]);
 118   double clock = smpi_process_simulated_elapsed();
 119 #ifdef HAVE_TRACING
 120   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 121   TRACE_smpi_computing_out(rank);
 122   int dst_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), to);
 123   TRACE_smpi_ptp_in(rank, rank, dst_traced, __FUNCTION__);
 124   TRACE_smpi_send(rank, rank, dst_traced);
 125 #endif
 126
 127   smpi_mpi_send(NULL, size, MPI_BYTE, to , 0, MPI_COMM_WORLD);
 128
 129   log_timed_action (action, clock);
 130
 131   #ifdef HAVE_TRACING
 132   TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
 133   TRACE_smpi_computing_in(rank);
 134 #endif
 135
 136 }
 137
 138 static void action_Isend(const char *const *action)
 139 {
 140   int to = atoi(action[2]);
 141   double size=parse_double(action[3]);
 142   double clock = smpi_process_simulated_elapsed();
 143   MPI_Request request;
 144   smpi_replay_globals_t globals =
 145      (smpi_replay_globals_t) smpi_process_get_user_data();
 146 #ifdef HAVE_TRACING
 147   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 148   TRACE_smpi_computing_out(rank);
 149   int dst_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), to);
 150   TRACE_smpi_ptp_in(rank, rank, dst_traced, __FUNCTION__);
 151   TRACE_smpi_send(rank, rank, dst_traced);
 152 #endif
 153
 154   request = smpi_mpi_isend(NULL, size, MPI_BYTE, to, 0,MPI_COMM_WORLD);
 155
 156 #ifdef HAVE_TRACING
 157   TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
 158   request->send = 1;
 159   TRACE_smpi_computing_in(rank);
 160 #endif
 161
 162   xbt_dynar_push(globals->isends,&request);
 163   xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
 164
 165   log_timed_action (action, clock);
 166 }
 167
 168 static void action_recv(const char *const *action) {
 169   int from = atoi(action[2]);
 170   double size=parse_double(action[3]);
 171   double clock = smpi_process_simulated_elapsed();
 172   MPI_Status status;
 173 #ifdef HAVE_TRACING
 174   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 175   int src_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), from);
 176   TRACE_smpi_computing_out(rank);
 177
 178   TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__);
 179 #endif
 180
 181   smpi_mpi_recv(NULL, size, MPI_BYTE, from, 0, MPI_COMM_WORLD, &status);
 182
 183 #ifdef HAVE_TRACING
 184   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
 185   TRACE_smpi_recv(rank, src_traced, rank);
 186   TRACE_smpi_computing_in(rank);
 187 #endif
 188
 189   log_timed_action (action, clock);
 190 }
 191
 192 static void action_Irecv(const char *const *action)
 193 {
 194   int from = atoi(action[2]);
 195   double size=parse_double(action[3]);
 196   double clock = smpi_process_simulated_elapsed();
 197   MPI_Request request;
 198   smpi_replay_globals_t globals =
 199      (smpi_replay_globals_t) smpi_process_get_user_data();
 200
 201 #ifdef HAVE_TRACING
 202   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 203   int src_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), from);
 204   TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__);
 205 #endif
 206
 207   request = smpi_mpi_irecv(NULL, size, MPI_BYTE, from, 0, MPI_COMM_WORLD);
 208
 209 #ifdef HAVE_TRACING
 210   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
 211   request->recv = 1;
 212 #endif
 213   xbt_dynar_push(globals->irecvs,&request);
 214   xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
 215
 216   log_timed_action (action, clock);
 217 }
 218
 219 static void action_wait(const char *const *action){
 220   double clock = smpi_process_simulated_elapsed();
 221   MPI_Request request;
 222   MPI_Status status;
 223   smpi_replay_globals_t globals =
 224       (smpi_replay_globals_t) smpi_process_get_user_data();
 225
 226   xbt_assert(xbt_dynar_length(globals->irecvs),
 227       "action wait not preceded by any irecv: %s",
 228       xbt_str_join_array(action," "));
 229   request = xbt_dynar_pop_as(globals->irecvs,MPI_Request);
 230 #ifdef HAVE_TRACING
 231   int rank = request && request->comm != MPI_COMM_NULL
 232       ? smpi_comm_rank(request->comm)
 233       : -1;
 234   TRACE_smpi_computing_out(rank);
 235
 236   MPI_Group group = smpi_comm_group(request->comm);
 237   int src_traced = smpi_group_rank(group, request->src);
 238   int dst_traced = smpi_group_rank(group, request->dst);
 239   int is_wait_for_receive = request->recv;
 240   TRACE_smpi_ptp_in(rank, src_traced, dst_traced, __FUNCTION__);
 241 #endif
 242   smpi_mpi_wait(&request, &status);
 243 #ifdef HAVE_TRACING
 244   TRACE_smpi_ptp_out(rank, src_traced, dst_traced, __FUNCTION__);
 245   if (is_wait_for_receive) {
 246     TRACE_smpi_recv(rank, src_traced, dst_traced);
 247   }
 248   TRACE_smpi_computing_in(rank);
 249 #endif
 250
 251   log_timed_action (action, clock);
 252 }
 253
 254 static void action_waitall(const char *const *action){
 255   double clock = smpi_process_simulated_elapsed();
 256   int count_requests=0;
 257   unsigned int i=0;
 258
 259   count_requests=xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 260
 261   if (count_requests>0) {
 262     MPI_Request requests[count_requests];
 263     MPI_Status status[count_requests];
 264
 265     /*  The reqq is an array of dynars. Its index corresponds to the rank.
 266      Thus each rank saves its own requests to the array request. */
 267     xbt_dynar_foreach(reqq[smpi_comm_rank(MPI_COMM_WORLD)],i,requests[i]);
 268
 269   #ifdef HAVE_TRACING
 270    //save information from requests
 271
 272    xbt_dynar_t srcs = xbt_dynar_new(sizeof(int), NULL);
 273    xbt_dynar_t dsts = xbt_dynar_new(sizeof(int), NULL);
 274    xbt_dynar_t recvs = xbt_dynar_new(sizeof(int), NULL);
 275    for (i = 0; i < count_requests; i++) {
 276     if(requests[i]){
 277       int *asrc = xbt_new(int, 1);
 278       int *adst = xbt_new(int, 1);
 279       int *arecv = xbt_new(int, 1);
 280       *asrc = requests[i]->src;
 281       *adst = requests[i]->dst;
 282       *arecv = requests[i]->recv;
 283       xbt_dynar_insert_at(srcs, i, asrc);
 284       xbt_dynar_insert_at(dsts, i, adst);
 285       xbt_dynar_insert_at(recvs, i, arecv);
 286       xbt_free(asrc);
 287       xbt_free(adst);
 288       xbt_free(arecv);
 289     }else {
 290       int *t = xbt_new(int, 1);
 291       xbt_dynar_insert_at(srcs, i, t);
 292       xbt_dynar_insert_at(dsts, i, t);
 293       xbt_dynar_insert_at(recvs, i, t);
 294       xbt_free(t);
 295     }
 296    }
 297    int rank_traced = smpi_process_index();
 298    TRACE_smpi_computing_out(rank_traced);
 299
 300    TRACE_smpi_ptp_in(rank_traced, -1, -1, __FUNCTION__);
 301   #endif
 302
 303     smpi_mpi_waitall(count_requests, requests, status);
 304
 305   #ifdef HAVE_TRACING
 306    for (i = 0; i < count_requests; i++) {
 307     int src_traced, dst_traced, is_wait_for_receive;
 308     xbt_dynar_get_cpy(srcs, i, &src_traced);
 309     xbt_dynar_get_cpy(dsts, i, &dst_traced);
 310     xbt_dynar_get_cpy(recvs, i, &is_wait_for_receive);
 311     if (is_wait_for_receive) {
 312       TRACE_smpi_recv(rank_traced, src_traced, dst_traced);
 313     }
 314    }
 315    TRACE_smpi_ptp_out(rank_traced, -1, -1, __FUNCTION__);
 316    //clean-up of dynars
 317    xbt_dynar_free(&srcs);
 318    xbt_dynar_free(&dsts);
 319    xbt_dynar_free(&recvs);
 320    TRACE_smpi_computing_in(rank_traced);
 321   #endif
 322
 323    xbt_dynar_reset(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 324   }
 325   log_timed_action (action, clock);
 326 }
 327
 328 static void action_barrier(const char *const *action){
 329   double clock = smpi_process_simulated_elapsed();
 330 #ifdef HAVE_TRACING
 331   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 332   TRACE_smpi_computing_out(rank);
 333   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 334 #endif
 335   smpi_mpi_barrier(MPI_COMM_WORLD);
 336 #ifdef HAVE_TRACING
 337   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 338   TRACE_smpi_computing_in(rank);
 339 #endif
 340
 341   log_timed_action (action, clock);
 342 }
 343
 344 static void action_bcast(const char *const *action)
 345 {
 346   double size = parse_double(action[2]);
 347   double clock = smpi_process_simulated_elapsed();
 348 #ifdef HAVE_TRACING
 349   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 350   TRACE_smpi_computing_out(rank);
 351   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), 0);
 352   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__);
 353 #endif
 354
 355   smpi_mpi_bcast(NULL, size, MPI_BYTE, 0, MPI_COMM_WORLD);
 356 #ifdef HAVE_TRACING
 357   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 358   TRACE_smpi_computing_in(rank);
 359 #endif
 360
 361   log_timed_action (action, clock);
 362 }
 363
 364 static void action_reduce(const char *const *action)
 365 {
 366   double size = parse_double(action[2]);
 367   double clock = smpi_process_simulated_elapsed();
 368 #ifdef HAVE_TRACING
 369   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 370   TRACE_smpi_computing_out(rank);
 371   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), 0);
 372   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__);
 373 #endif
 374    smpi_mpi_reduce(NULL, NULL, size, MPI_BYTE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
 375 #ifdef HAVE_TRACING
 376   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 377   TRACE_smpi_computing_in(rank);
 378 #endif
 379
 380   log_timed_action (action, clock);
 381 }
 382
 383 static void action_allReduce(const char *const *action) {
 384   double comm_size = parse_double(action[2]);
 385   double comp_size = parse_double(action[3]);
 386   double clock = smpi_process_simulated_elapsed();
 387 #ifdef HAVE_TRACING
 388   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 389   TRACE_smpi_computing_out(rank);
 390   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 391 #endif
 392   smpi_mpi_reduce(NULL, NULL, comm_size, MPI_BYTE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
 393   smpi_execute_flops(comp_size);
 394   smpi_mpi_bcast(NULL, comm_size, MPI_BYTE, 0, MPI_COMM_WORLD);
 395 #ifdef HAVE_TRACING
 396   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 397   TRACE_smpi_computing_in(rank);
 398 #endif
 399
 400   log_timed_action (action, clock);
 401 }
 402
 403 static void action_allToAll(const char *const *action) {
 404   double clock = smpi_process_simulated_elapsed();
 405   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 406   int send_size = atoi(action[2]);
 407   int recv_size = atoi(action[3]);
 408   void *send = xbt_new0(int, send_size*comm_size);
 409   void *recv = xbt_new0(int, send_size*comm_size);
 410
 411
 412 #ifdef HAVE_TRACING
 413   int rank = smpi_process_index();
 414   TRACE_smpi_computing_out(rank);
 415   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 416 #endif
 417
 418   if (send_size < 200 && comm_size > 12) {
 419     smpi_coll_tuned_alltoall_bruck(send, send_size, MPI_BYTE,
 420                                    recv, recv_size, MPI_BYTE,
 421                                    MPI_COMM_WORLD);
 422   } else if (send_size < 3000) {
 423
 424     smpi_coll_tuned_alltoall_basic_linear(send, send_size, MPI_BYTE,
 425                                           recv, recv_size, MPI_BYTE,
 426                                           MPI_COMM_WORLD);
 427   } else {
 428     smpi_coll_tuned_alltoall_pairwise(send, send_size, MPI_BYTE,
 429                                       recv, recv_size, MPI_BYTE,
 430                                       MPI_COMM_WORLD);
 431   }
 432
 433 #ifdef HAVE_TRACING
 434   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 435   TRACE_smpi_computing_in(rank);
 436 #endif
 437
 438   log_timed_action (action, clock);
 439   xbt_free(send);
 440   xbt_free(recv);
 441 }
 442
 443 static void action_allToAllv(const char *const *action) {
 444   double clock = smpi_process_simulated_elapsed();
 445 //  int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 446
 447 //  PMPI_Alltoallv(NULL, send_size, send_disp,
 448 //                   MPI_BYTE, NULL, recv_size,
 449 //                   recv_disp, MPI_BYTE, MPI_COMM_WORLD);
 450
 451
 452   log_timed_action (action, clock);
 453
 454 }
 455
 456 void smpi_replay_init(int *argc, char***argv){
 457   PMPI_Init(argc, argv);
 458   if (!smpi_process_index()){
 459     _xbt_replay_action_init();
 460     xbt_replay_action_register("init",       action_init);
 461     xbt_replay_action_register("finalize",   action_finalize);
 462     xbt_replay_action_register("comm_size",  action_comm_size);
 463     xbt_replay_action_register("comm_split", action_comm_split);
 464     xbt_replay_action_register("comm_dup",   action_comm_dup);
 465     xbt_replay_action_register("send",       action_send);
 466     xbt_replay_action_register("Isend",      action_Isend);
 467     xbt_replay_action_register("recv",       action_recv);
 468     xbt_replay_action_register("Irecv",      action_Irecv);
 469     xbt_replay_action_register("wait",       action_wait);
 470     xbt_replay_action_register("waitAll",    action_waitall);
 471     xbt_replay_action_register("barrier",    action_barrier);
 472     xbt_replay_action_register("bcast",      action_bcast);
 473     xbt_replay_action_register("reduce",     action_reduce);
 474     xbt_replay_action_register("allReduce",  action_allReduce);
 475     xbt_replay_action_register("allToAll",   action_allToAll);
 476     xbt_replay_action_register("allToAllV",  action_allToAllv);
 477     xbt_replay_action_register("compute",    action_compute);
 478   }
 479
 480   xbt_replay_action_runner(*argc, *argv);
 481 }
 482
 483 int smpi_replay_finalize(){
 484   double sim_time= 1.;
 485   /* One active process will stop. Decrease the counter*/
 486   active_processes--;
 487   if(!active_processes){
 488     /* Last process alive speaking */
 489     /* end the simulated timer */
 490     xbt_dynar_free(reqq);
 491     sim_time = smpi_process_simulated_elapsed();
 492     XBT_INFO("Simulation time %g", sim_time);
 493     _xbt_replay_action_exit();
 494   }
 495   return PMPI_Finalize();
 496 }