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,req_counts=0,i=0;
 257     smpi_replay_globals_t globals =
 258       (smpi_replay_globals_t) smpi_process_get_user_data();
 259
 260   count_requests=xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 261
 262   if (count_requests>0) {
 263     MPI_Request requests[count_requests];
 264     MPI_Status status[count_requests];
 265
 266     for(i=0;i<count_requests;i++){
 267       xbt_dynar_foreach(reqq[smpi_comm_rank(MPI_COMM_WORLD)],i,requests[i]);
 268     }
 269
 270   #ifdef HAVE_TRACING
 271    //save information from requests
 272
 273    xbt_dynar_t srcs = xbt_dynar_new(sizeof(int), NULL);
 274    xbt_dynar_t dsts = xbt_dynar_new(sizeof(int), NULL);
 275    xbt_dynar_t recvs = xbt_dynar_new(sizeof(int), NULL);
 276    for (i = 0; i < count_requests; i++) {
 277     if(requests[i]){
 278       int *asrc = xbt_new(int, 1);
 279       int *adst = xbt_new(int, 1);
 280       int *arecv = xbt_new(int, 1);
 281       *asrc = requests[i]->src;
 282       *adst = requests[i]->dst;
 283       *arecv = requests[i]->recv;
 284       xbt_dynar_insert_at(srcs, i, asrc);
 285       xbt_dynar_insert_at(dsts, i, adst);
 286       xbt_dynar_insert_at(recvs, i, arecv);
 287       xbt_free(asrc);
 288       xbt_free(adst);
 289       xbt_free(arecv);
 290     }else {
 291       int *t = xbt_new(int, 1);
 292       xbt_dynar_insert_at(srcs, i, t);
 293       xbt_dynar_insert_at(dsts, i, t);
 294       xbt_dynar_insert_at(recvs, i, t);
 295       xbt_free(t);
 296     }
 297    }
 298    int rank_traced = smpi_process_index();
 299    TRACE_smpi_computing_out(rank_traced);
 300
 301    TRACE_smpi_ptp_in(rank_traced, -1, -1, __FUNCTION__);
 302   #endif
 303
 304     smpi_mpi_waitall(count_requests, requests, status);
 305
 306   #ifdef HAVE_TRACING
 307    for (i = 0; i < count_requests; i++) {
 308     int src_traced, dst_traced, is_wait_for_receive;
 309     xbt_dynar_get_cpy(srcs, i, &src_traced);
 310     xbt_dynar_get_cpy(dsts, i, &dst_traced);
 311     xbt_dynar_get_cpy(recvs, i, &is_wait_for_receive);
 312     if (is_wait_for_receive) {
 313       TRACE_smpi_recv(rank_traced, src_traced, dst_traced);
 314     }
 315    }
 316    TRACE_smpi_ptp_out(rank_traced, -1, -1, __FUNCTION__);
 317    //clean-up of dynars
 318    xbt_dynar_free(&srcs);
 319    xbt_dynar_free(&dsts);
 320    xbt_dynar_free(&recvs);
 321    TRACE_smpi_computing_in(rank_traced);
 322   #endif
 323
 324    xbt_dynar_reset(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 325   }
 326   log_timed_action (action, clock);
 327 }
 328
 329 static void action_barrier(const char *const *action){
 330   double clock = smpi_process_simulated_elapsed();
 331 #ifdef HAVE_TRACING
 332   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 333   TRACE_smpi_computing_out(rank);
 334   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 335 #endif
 336   smpi_mpi_barrier(MPI_COMM_WORLD);
 337 #ifdef HAVE_TRACING
 338   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 339   TRACE_smpi_computing_in(rank);
 340 #endif
 341
 342   log_timed_action (action, clock);
 343 }
 344
 345 static void action_bcast(const char *const *action)
 346 {
 347   double size = parse_double(action[2]);
 348   double clock = smpi_process_simulated_elapsed();
 349 #ifdef HAVE_TRACING
 350   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 351   TRACE_smpi_computing_out(rank);
 352   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), 0);
 353   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__);
 354 #endif
 355
 356   smpi_mpi_bcast(NULL, size, MPI_BYTE, 0, MPI_COMM_WORLD);
 357 #ifdef HAVE_TRACING
 358   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 359   TRACE_smpi_computing_in(rank);
 360 #endif
 361
 362   log_timed_action (action, clock);
 363 }
 364
 365 static void action_reduce(const char *const *action)
 366 {
 367   double size = parse_double(action[2]);
 368   double clock = smpi_process_simulated_elapsed();
 369 #ifdef HAVE_TRACING
 370   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 371   TRACE_smpi_computing_out(rank);
 372   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), 0);
 373   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__);
 374 #endif
 375    smpi_mpi_reduce(NULL, NULL, size, MPI_BYTE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
 376 #ifdef HAVE_TRACING
 377   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 378   TRACE_smpi_computing_in(rank);
 379 #endif
 380
 381   log_timed_action (action, clock);
 382 }
 383
 384 static void action_allReduce(const char *const *action) {
 385   double comm_size = parse_double(action[2]);
 386   double comp_size = parse_double(action[3]);
 387   double clock = smpi_process_simulated_elapsed();
 388 #ifdef HAVE_TRACING
 389   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 390   TRACE_smpi_computing_out(rank);
 391   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 392 #endif
 393   smpi_mpi_reduce(NULL, NULL, comm_size, MPI_BYTE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
 394   smpi_execute_flops(comp_size);
 395   smpi_mpi_bcast(NULL, comm_size, MPI_BYTE, 0, MPI_COMM_WORLD);
 396 #ifdef HAVE_TRACING
 397   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 398   TRACE_smpi_computing_in(rank);
 399 #endif
 400
 401   log_timed_action (action, clock);
 402 }
 403
 404 static void action_allToAll(const char *const *action) {
 405   double clock = smpi_process_simulated_elapsed();
 406   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 407   int send_size = atoi(action[2]);
 408   int recv_size = atoi(action[3]);
 409   void *send = xbt_new0(int, send_size*comm_size);
 410   void *recv = xbt_new0(int, send_size*comm_size);
 411
 412
 413 #ifdef HAVE_TRACING
 414   int rank = smpi_process_index();
 415   TRACE_smpi_computing_out(rank);
 416   TRACE_smpi_collective_in(rank, -1, __FUNCTION__);
 417 #endif
 418
 419   if (send_size < 200 && comm_size > 12) {
 420     smpi_coll_tuned_alltoall_bruck(send, send_size, MPI_BYTE,
 421                                    recv, recv_size, MPI_BYTE,
 422                                    MPI_COMM_WORLD);
 423   } else if (send_size < 3000) {
 424
 425     smpi_coll_tuned_alltoall_basic_linear(send, send_size, MPI_BYTE,
 426                                           recv, recv_size, MPI_BYTE,
 427                                           MPI_COMM_WORLD);
 428   } else {
 429     smpi_coll_tuned_alltoall_pairwise(send, send_size, MPI_BYTE,
 430                                       recv, recv_size, MPI_BYTE,
 431                                       MPI_COMM_WORLD);
 432   }
 433
 434 #ifdef HAVE_TRACING
 435   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 436   TRACE_smpi_computing_in(rank);
 437 #endif
 438
 439   log_timed_action (action, clock);
 440   xbt_free(send);
 441   xbt_free(recv);
 442 }
 443
 444 static void action_allToAllv(const char *const *action) {
 445   double clock = smpi_process_simulated_elapsed();
 446   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 447
 448 //  PMPI_Alltoallv(NULL, send_size, send_disp,
 449 //                   MPI_BYTE, NULL, recv_size,
 450 //                   recv_disp, MPI_BYTE, MPI_COMM_WORLD);
 451
 452
 453   log_timed_action (action, clock);
 454
 455 }
 456
 457 void smpi_replay_init(int *argc, char***argv){
 458   PMPI_Init(argc, argv);
 459   if (!smpi_process_index()){
 460     _xbt_replay_action_init();
 461     xbt_replay_action_register("init",       action_init);
 462     xbt_replay_action_register("finalize",   action_finalize);
 463     xbt_replay_action_register("comm_size",  action_comm_size);
 464     xbt_replay_action_register("comm_split", action_comm_split);
 465     xbt_replay_action_register("comm_dup",   action_comm_dup);
 466     xbt_replay_action_register("send",       action_send);
 467     xbt_replay_action_register("Isend",      action_Isend);
 468     xbt_replay_action_register("recv",       action_recv);
 469     xbt_replay_action_register("Irecv",      action_Irecv);
 470     xbt_replay_action_register("wait",       action_wait);
 471     xbt_replay_action_register("waitAll",    action_waitall);
 472     xbt_replay_action_register("barrier",    action_barrier);
 473     xbt_replay_action_register("bcast",      action_bcast);
 474     xbt_replay_action_register("reduce",     action_reduce);
 475     xbt_replay_action_register("allReduce",  action_allReduce);
 476     xbt_replay_action_register("allToAll",   action_allToAll);
 477     xbt_replay_action_register("allToAllV",  action_allToAllv);
 478     xbt_replay_action_register("compute",    action_compute);
 479   }
 480
 481   xbt_replay_action_runner(*argc, *argv);
 482 }
 483
 484 int smpi_replay_finalize(){
 485   double sim_time= 1.;
 486   /* One active process will stop. Decrease the counter*/
 487   active_processes--;
 488   if(!active_processes){
 489     /* Last process alive speaking */
 490     /* end the simulated timer */
 491     xbt_dynar_free(reqq);
 492     sim_time = smpi_process_simulated_elapsed();
 493     XBT_INFO("Simulation time %g", sim_time);
 494     _xbt_replay_action_exit();
 495   }
 496   return PMPI_Finalize();
 497 }