src/smpi/internals/smpi_replay.cpp

   1 /* Copyright (c) 2009-2019. The SimGrid Team. All rights reserved.          */
   2
   3 /* This program is free software; you can redistribute it and/or modify it
   4  * under the terms of the license (GNU LGPL) which comes with this package. */
   5
   6 #include "smpi_coll.hpp"
   7 #include "smpi_comm.hpp"
   8 #include "smpi_datatype.hpp"
   9 #include "smpi_group.hpp"
  10 #include "smpi_request.hpp"
  11 #include "xbt/replay.hpp"
  12 #include <simgrid/smpi/replay.hpp>
  13 #include <src/smpi/include/private.hpp>
  14
  15 #include <memory>
  16 #include <numeric>
  17 #include <unordered_map>
  18 #include <vector>
  19
  20 #include <tuple>
  21
  22 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_replay, smpi, "Trace Replay with SMPI");
  23
  24 // From https://stackoverflow.com/questions/7110301/generic-hash-for-tuples-in-unordered-map-unordered-set
  25 // This is all just to make std::unordered_map work with std::tuple. If we need this in other places,
  26 // this could go into a header file.
  27 namespace hash_tuple {
  28 template <typename TT> class hash {
  29 public:
  30   size_t operator()(TT const& tt) const { return std::hash<TT>()(tt); }
  31 };
  32
  33 template <class T> inline void hash_combine(std::size_t& seed, T const& v)
  34 {
  35   seed ^= hash_tuple::hash<T>()(v) + 0x9e3779b9 + (seed << 6) + (seed >> 2);
  36 }
  37
  38 // Recursive template code derived from Matthieu M.
  39 template <class Tuple, size_t Index = std::tuple_size<Tuple>::value - 1> class HashValueImpl {
  40 public:
  41   static void apply(size_t& seed, Tuple const& tuple)
  42   {
  43     HashValueImpl<Tuple, Index - 1>::apply(seed, tuple);
  44     hash_combine(seed, std::get<Index>(tuple));
  45   }
  46 };
  47
  48 template <class Tuple> class HashValueImpl<Tuple, 0> {
  49 public:
  50   static void apply(size_t& seed, Tuple const& tuple) { hash_combine(seed, std::get<0>(tuple)); }
  51 };
  52
  53 template <typename... TT> class hash<std::tuple<TT...>> {
  54 public:
  55   size_t operator()(std::tuple<TT...> const& tt) const
  56   {
  57     size_t seed = 0;
  58     HashValueImpl<std::tuple<TT...>>::apply(seed, tt);
  59     return seed;
  60   }
  61 };
  62 }
  63
  64 typedef std::tuple</*sender*/ int, /* reciever */ int, /* tag */int> req_key_t;
  65 typedef std::unordered_map<req_key_t, MPI_Request, hash_tuple::hash<std::tuple<int,int,int>>> req_storage_t;
  66
  67 void log_timed_action(simgrid::xbt::ReplayAction& action, double clock)
  68 {
  69   if (XBT_LOG_ISENABLED(smpi_replay, xbt_log_priority_verbose)){
  70     std::string s = boost::algorithm::join(action, " ");
  71     XBT_VERB("%s %f", s.c_str(), smpi_process()->simulated_elapsed() - clock);
  72   }
  73 }
  74
  75 /* Helper function */
  76 static double parse_double(const std::string& string)
  77 {
  78   return xbt_str_parse_double(string.c_str(), "%s is not a double");
  79 }
  80
  81 namespace simgrid {
  82 namespace smpi {
  83
  84 namespace replay {
  85 MPI_Datatype MPI_DEFAULT_TYPE;
  86
  87 class RequestStorage {
  88 private:
  89     req_storage_t store;
  90
  91 public:
  92     RequestStorage() {}
  93     int size()
  94     {
  95       return store.size();
  96     }
  97
  98     req_storage_t& get_store()
  99     {
 100       return store;
 101     }
 102
 103     void get_requests(std::vector<MPI_Request>& vec)
 104     {
 105       for (auto& pair : store) {
 106         auto& req = pair.second;
 107         auto my_proc_id = simgrid::s4u::this_actor::get_pid();
 108         if (req != MPI_REQUEST_NULL && (req->src() == my_proc_id || req->dst() == my_proc_id)) {
 109           vec.push_back(pair.second);
 110           pair.second->print_request("MM");
 111         }
 112       }
 113     }
 114
 115     MPI_Request find(int src, int dst, int tag)
 116     {
 117       req_storage_t::iterator it = store.find(req_key_t(src, dst, tag));
 118       return (it == store.end()) ? MPI_REQUEST_NULL : it->second;
 119     }
 120
 121     void remove(MPI_Request req)
 122     {
 123       if (req == MPI_REQUEST_NULL) return;
 124
 125       store.erase(req_key_t(req->src()-1, req->dst()-1, req->tag()));
 126     }
 127
 128     void add(MPI_Request req)
 129     {
 130       if (req != MPI_REQUEST_NULL) // Can and does happen in the case of TestAction
 131         store.insert({req_key_t(req->src()-1, req->dst()-1, req->tag()), req});
 132     }
 133
 134     /* Sometimes we need to re-insert MPI_REQUEST_NULL but we still need src,dst and tag */
 135     void addNullRequest(int src, int dst, int tag)
 136     {
 137       store.insert({req_key_t(
 138             MPI_COMM_WORLD->group()->actor(src)->get_pid()-1,
 139             MPI_COMM_WORLD->group()->actor(dst)->get_pid()-1,
 140             tag), MPI_REQUEST_NULL});
 141     }
 142 };
 143
 144 void WaitTestParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 145 {
 146   CHECK_ACTION_PARAMS(action, 3, 0)
 147   src = std::stoi(action[2]);
 148   dst = std::stoi(action[3]);
 149   tag = std::stoi(action[4]);
 150 }
 151
 152 void SendRecvParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 153 {
 154   CHECK_ACTION_PARAMS(action, 3, 1)
 155   partner = std::stoi(action[2]);
 156   tag     = std::stoi(action[3]);
 157   size    = parse_double(action[4]);
 158   if (action.size() > 5)
 159     datatype1 = simgrid::smpi::Datatype::decode(action[5]);
 160 }
 161
 162 void ComputeParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 163 {
 164   CHECK_ACTION_PARAMS(action, 1, 0)
 165   flops = parse_double(action[2]);
 166 }
 167
 168 void BcastArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 169 {
 170   CHECK_ACTION_PARAMS(action, 1, 2)
 171   size = parse_double(action[2]);
 172   root = (action.size() > 3) ? std::stoi(action[3]) : 0;
 173   if (action.size() > 4)
 174     datatype1 = simgrid::smpi::Datatype::decode(action[4]);
 175 }
 176
 177 void ReduceArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 178 {
 179   CHECK_ACTION_PARAMS(action, 2, 2)
 180   comm_size = parse_double(action[2]);
 181   comp_size = parse_double(action[3]);
 182   root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
 183   if (action.size() > 5)
 184     datatype1 = simgrid::smpi::Datatype::decode(action[5]);
 185 }
 186
 187 void AllReduceArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 188 {
 189   CHECK_ACTION_PARAMS(action, 2, 1)
 190   comm_size = parse_double(action[2]);
 191   comp_size = parse_double(action[3]);
 192   if (action.size() > 4)
 193     datatype1 = simgrid::smpi::Datatype::decode(action[4]);
 194 }
 195
 196 void AllToAllArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 197 {
 198   CHECK_ACTION_PARAMS(action, 2, 1)
 199   comm_size = MPI_COMM_WORLD->size();
 200   send_size = parse_double(action[2]);
 201   recv_size = parse_double(action[3]);
 202
 203   if (action.size() > 4)
 204     datatype1 = simgrid::smpi::Datatype::decode(action[4]);
 205   if (action.size() > 5)
 206     datatype2 = simgrid::smpi::Datatype::decode(action[5]);
 207 }
 208
 209 void GatherArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string& name)
 210 {
 211   /* The structure of the gather action for the rank 0 (total 4 processes) is the following:
 212         0 gather 68 68 0 0 0
 213       where:
 214         1) 68 is the sendcounts
 215         2) 68 is the recvcounts
 216         3) 0 is the root node
 217         4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
 218         5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
 219   */
 220   CHECK_ACTION_PARAMS(action, 2, 3)
 221   comm_size = MPI_COMM_WORLD->size();
 222   send_size = parse_double(action[2]);
 223   recv_size = parse_double(action[3]);
 224
 225   if (name == "gather") {
 226     root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
 227     if (action.size() > 5)
 228       datatype1 = simgrid::smpi::Datatype::decode(action[5]);
 229     if (action.size() > 6)
 230       datatype2 = simgrid::smpi::Datatype::decode(action[6]);
 231   } else {
 232     if (action.size() > 4)
 233       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
 234     if (action.size() > 5)
 235       datatype2 = simgrid::smpi::Datatype::decode(action[5]);
 236   }
 237 }
 238
 239 void GatherVArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string& name)
 240 {
 241   /* The structure of the gatherv action for the rank 0 (total 4 processes) is the following:
 242        0 gather 68 68 10 10 10 0 0 0
 243      where:
 244        1) 68 is the sendcount
 245        2) 68 10 10 10 is the recvcounts
 246        3) 0 is the root node
 247        4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
 248        5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
 249   */
 250   comm_size = MPI_COMM_WORLD->size();
 251   CHECK_ACTION_PARAMS(action, comm_size + 1, 2)
 252   send_size  = parse_double(action[2]);
 253   disps      = std::vector<int>(comm_size, 0);
 254   recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
 255
 256   if (name == "gatherv") {
 257     root = (action.size() > 3 + comm_size) ? std::stoi(action[3 + comm_size]) : 0;
 258     if (action.size() > 4 + comm_size)
 259       datatype1 = simgrid::smpi::Datatype::decode(action[4 + comm_size]);
 260     if (action.size() > 5 + comm_size)
 261       datatype2 = simgrid::smpi::Datatype::decode(action[5 + comm_size]);
 262   } else {
 263     int datatype_index = 0;
 264     int disp_index     = 0;
 265     /* The 3 comes from "0 gather <sendcount>", which must always be present.
 266      * The + comm_size is the recvcounts array, which must also be present
 267      */
 268     if (action.size() > 3 + comm_size + comm_size) { /* datatype + disp are specified */
 269       datatype_index = 3 + comm_size;
 270       disp_index     = datatype_index + 1;
 271       datatype1      = simgrid::smpi::Datatype::decode(action[datatype_index]);
 272       datatype2      = simgrid::smpi::Datatype::decode(action[datatype_index]);
 273     } else if (action.size() >
 274                3 + comm_size + 2) { /* disps specified; datatype is not specified; use the default one */
 275       disp_index = 3 + comm_size;
 276     } else if (action.size() > 3 + comm_size) { /* only datatype, no disp specified */
 277       datatype_index = 3 + comm_size;
 278       datatype1      = simgrid::smpi::Datatype::decode(action[datatype_index]);
 279       datatype2      = simgrid::smpi::Datatype::decode(action[datatype_index]);
 280     }
 281
 282     if (disp_index != 0) {
 283       for (unsigned int i = 0; i < comm_size; i++)
 284         disps[i]          = std::stoi(action[disp_index + i]);
 285     }
 286   }
 287
 288   for (unsigned int i = 0; i < comm_size; i++) {
 289     (*recvcounts)[i] = std::stoi(action[i + 3]);
 290   }
 291   recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
 292 }
 293
 294 void ScatterArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 295 {
 296   /* The structure of the scatter action for the rank 0 (total 4 processes) is the following:
 297         0 gather 68 68 0 0 0
 298       where:
 299         1) 68 is the sendcounts
 300         2) 68 is the recvcounts
 301         3) 0 is the root node
 302         4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
 303         5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
 304   */
 305   CHECK_ACTION_PARAMS(action, 2, 3)
 306   comm_size = MPI_COMM_WORLD->size();
 307   send_size = parse_double(action[2]);
 308   recv_size = parse_double(action[3]);
 309   root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
 310   if (action.size() > 5)
 311     datatype1 = simgrid::smpi::Datatype::decode(action[5]);
 312   if (action.size() > 6)
 313     datatype2 = simgrid::smpi::Datatype::decode(action[6]);
 314 }
 315
 316 void ScatterVArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 317 {
 318   /* The structure of the scatterv action for the rank 0 (total 4 processes) is the following:
 319      0 gather 68 10 10 10 68 0 0 0
 320       where:
 321       1) 68 10 10 10 is the sendcounts
 322       2) 68 is the recvcount
 323       3) 0 is the root node
 324       4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
 325       5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
 326   */
 327   CHECK_ACTION_PARAMS(action, comm_size + 1, 2)
 328   recv_size  = parse_double(action[2 + comm_size]);
 329   disps      = std::vector<int>(comm_size, 0);
 330   sendcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
 331
 332   if (action.size() > 5 + comm_size)
 333     datatype1 = simgrid::smpi::Datatype::decode(action[4 + comm_size]);
 334   if (action.size() > 5 + comm_size)
 335     datatype2 = simgrid::smpi::Datatype::decode(action[5]);
 336
 337   for (unsigned int i = 0; i < comm_size; i++) {
 338     (*sendcounts)[i] = std::stoi(action[i + 2]);
 339   }
 340   send_size_sum = std::accumulate(sendcounts->begin(), sendcounts->end(), 0);
 341   root          = (action.size() > 3 + comm_size) ? std::stoi(action[3 + comm_size]) : 0;
 342 }
 343
 344 void ReduceScatterArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 345 {
 346   /* The structure of the reducescatter action for the rank 0 (total 4 processes) is the following:
 347        0 reducescatter 275427 275427 275427 204020 11346849 0
 348      where:
 349        1) The first four values after the name of the action declare the recvcounts array
 350        2) The value 11346849 is the amount of instructions
 351        3) The last value corresponds to the datatype, see simgrid::smpi::Datatype::decode().
 352   */
 353   comm_size = MPI_COMM_WORLD->size();
 354   CHECK_ACTION_PARAMS(action, comm_size + 1, 1)
 355   comp_size  = parse_double(action[2 + comm_size]);
 356   recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
 357   if (action.size() > 3 + comm_size)
 358     datatype1 = simgrid::smpi::Datatype::decode(action[3 + comm_size]);
 359
 360   for (unsigned int i = 0; i < comm_size; i++) {
 361     recvcounts->push_back(std::stoi(action[i + 2]));
 362   }
 363   recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
 364 }
 365
 366 void AllToAllVArgParser::parse(simgrid::xbt::ReplayAction& action, const std::string&)
 367 {
 368   /* The structure of the alltoallv action for the rank 0 (total 4 processes) is the following:
 369         0 alltoallv 100 1 7 10 12 100 1 70 10 5
 370      where:
 371       1) 100 is the size of the send buffer *sizeof(int),
 372       2) 1 7 10 12 is the sendcounts array
 373       3) 100*sizeof(int) is the size of the receiver buffer
 374       4)  1 70 10 5 is the recvcounts array
 375   */
 376   comm_size = MPI_COMM_WORLD->size();
 377   CHECK_ACTION_PARAMS(action, 2 * comm_size + 2, 2)
 378   sendcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
 379   recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
 380   senddisps  = std::vector<int>(comm_size, 0);
 381   recvdisps  = std::vector<int>(comm_size, 0);
 382
 383   if (action.size() > 5 + 2 * comm_size)
 384     datatype1 = simgrid::smpi::Datatype::decode(action[4 + 2 * comm_size]);
 385   if (action.size() > 5 + 2 * comm_size)
 386     datatype2 = simgrid::smpi::Datatype::decode(action[5 + 2 * comm_size]);
 387
 388   send_buf_size = parse_double(action[2]);
 389   recv_buf_size = parse_double(action[3 + comm_size]);
 390   for (unsigned int i = 0; i < comm_size; i++) {
 391     (*sendcounts)[i] = std::stoi(action[3 + i]);
 392     (*recvcounts)[i] = std::stoi(action[4 + comm_size + i]);
 393   }
 394   send_size_sum = std::accumulate(sendcounts->begin(), sendcounts->end(), 0);
 395   recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
 396 }
 397
 398 void WaitAction::kernel(simgrid::xbt::ReplayAction& action)
 399 {
 400   std::string s = boost::algorithm::join(action, " ");
 401   xbt_assert(req_storage.size(), "action wait not preceded by any irecv or isend: %s", s.c_str());
 402   MPI_Request request = req_storage.find(args.src, args.dst, args.tag);
 403   req_storage.remove(request);
 404
 405   if (request == MPI_REQUEST_NULL) {
 406     /* Assume that the trace is well formed, meaning the comm might have been caught by a MPI_test. Then just
 407      * return.*/
 408     return;
 409   }
 410
 411   int rank = request->comm() != MPI_COMM_NULL ? request->comm()->rank() : -1;
 412
 413   // Must be taken before Request::wait() since the request may be set to
 414   // MPI_REQUEST_NULL by Request::wait!
 415   bool is_wait_for_receive = (request->flags() & MPI_REQ_RECV);
 416   // TODO: Here we take the rank while we normally take the process id (look for my_proc_id)
 417   TRACE_smpi_comm_in(rank, __func__, new simgrid::instr::WaitTIData(args.src, args.dst, args.tag));
 418
 419   MPI_Status status;
 420   Request::wait(&request, &status);
 421
 422   TRACE_smpi_comm_out(rank);
 423   if (is_wait_for_receive)
 424     TRACE_smpi_recv(args.src, args.dst, args.tag);
 425 }
 426
 427 void SendAction::kernel(simgrid::xbt::ReplayAction&)
 428 {
 429   int dst_traced = MPI_COMM_WORLD->group()->actor(args.partner)->get_pid();
 430
 431   TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData(name, args.partner, args.size,
 432         args.tag, Datatype::encode(args.datatype1)));
 433   if (not TRACE_smpi_view_internals())
 434     TRACE_smpi_send(my_proc_id, my_proc_id, dst_traced, args.tag, args.size * args.datatype1->size());
 435
 436   if (name == "send") {
 437     Request::send(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
 438   } else if (name == "isend") {
 439     MPI_Request request = Request::isend(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
 440     req_storage.add(request);
 441   } else {
 442     xbt_die("Don't know this action, %s", name.c_str());
 443   }
 444
 445   TRACE_smpi_comm_out(my_proc_id);
 446 }
 447
 448 void RecvAction::kernel(simgrid::xbt::ReplayAction&)
 449 {
 450   TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData(name, args.partner, args.size,
 451         args.tag, Datatype::encode(args.datatype1)));
 452
 453   MPI_Status status;
 454   // unknown size from the receiver point of view
 455   if (args.size <= 0.0) {
 456     Request::probe(args.partner, args.tag, MPI_COMM_WORLD, &status);
 457     args.size = status.count;
 458   }
 459
 460   bool is_recv = false; // Help analyzers understanding that status is not used unintialized
 461   if (name == "recv") {
 462     is_recv = true;
 463     Request::recv(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD, &status);
 464   } else if (name == "irecv") {
 465     MPI_Request request = Request::irecv(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
 466     req_storage.add(request);
 467   } else {
 468     THROW_IMPOSSIBLE;
 469   }
 470
 471   TRACE_smpi_comm_out(my_proc_id);
 472   if (is_recv && not TRACE_smpi_view_internals()) {
 473     int src_traced = MPI_COMM_WORLD->group()->actor(status.MPI_SOURCE)->get_pid();
 474     TRACE_smpi_recv(src_traced, my_proc_id, args.tag);
 475   }
 476 }
 477
 478 void ComputeAction::kernel(simgrid::xbt::ReplayAction&)
 479 {
 480   TRACE_smpi_computing_in(my_proc_id, args.flops);
 481   smpi_execute_flops(args.flops);
 482   TRACE_smpi_computing_out(my_proc_id);
 483 }
 484
 485 void TestAction::kernel(simgrid::xbt::ReplayAction&)
 486 {
 487   MPI_Request request = req_storage.find(args.src, args.dst, args.tag);
 488   req_storage.remove(request);
 489   // if request is null here, this may mean that a previous test has succeeded
 490   // Different times in traced application and replayed version may lead to this
 491   // In this case, ignore the extra calls.
 492   if (request != MPI_REQUEST_NULL) {
 493     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::NoOpTIData("test"));
 494
 495     MPI_Status status;
 496     int flag = Request::test(&request, &status);
 497
 498     XBT_DEBUG("MPI_Test result: %d", flag);
 499     /* push back request in vector to be caught by a subsequent wait. if the test did succeed, the request is now
 500      * nullptr.*/
 501     if (request == MPI_REQUEST_NULL)
 502       req_storage.addNullRequest(args.src, args.dst, args.tag);
 503     else
 504       req_storage.add(request);
 505
 506     TRACE_smpi_comm_out(my_proc_id);
 507   }
 508 }
 509
 510 void InitAction::kernel(simgrid::xbt::ReplayAction& action)
 511 {
 512   CHECK_ACTION_PARAMS(action, 0, 1)
 513     MPI_DEFAULT_TYPE = (action.size() > 2) ? MPI_DOUBLE // default MPE datatype
 514     : MPI_BYTE;  // default TAU datatype
 515
 516   /* start a simulated timer */
 517   smpi_process()->simulated_start();
 518 }
 519
 520 void CommunicatorAction::kernel(simgrid::xbt::ReplayAction&)
 521 {
 522   /* nothing to do */
 523 }
 524
 525 void WaitAllAction::kernel(simgrid::xbt::ReplayAction&)
 526 {
 527   const unsigned int count_requests = req_storage.size();
 528
 529   if (count_requests > 0) {
 530     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData("waitall", -1, count_requests, ""));
 531     std::vector<std::pair</*sender*/int,/*recv*/int>> sender_receiver;
 532     std::vector<MPI_Request> reqs;
 533     req_storage.get_requests(reqs);
 534     for (const auto& req : reqs) {
 535       if (req && (req->flags() & MPI_REQ_RECV)) {
 536         sender_receiver.push_back({req->src(), req->dst()});
 537       }
 538     }
 539     MPI_Status status[count_requests];
 540     Request::waitall(count_requests, &(reqs.data())[0], status);
 541     req_storage.get_store().clear();
 542
 543     for (auto& pair : sender_receiver) {
 544       TRACE_smpi_recv(pair.first, pair.second, 0);
 545     }
 546     TRACE_smpi_comm_out(my_proc_id);
 547   }
 548 }
 549
 550 void BarrierAction::kernel(simgrid::xbt::ReplayAction&)
 551 {
 552   TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::NoOpTIData("barrier"));
 553   Colls::barrier(MPI_COMM_WORLD);
 554   TRACE_smpi_comm_out(my_proc_id);
 555 }
 556
 557 void BcastAction::kernel(simgrid::xbt::ReplayAction&)
 558 {
 559   TRACE_smpi_comm_in(my_proc_id, "action_bcast",
 560       new simgrid::instr::CollTIData("bcast", MPI_COMM_WORLD->group()->actor(args.root)->get_pid(),
 561         -1.0, args.size, -1, Datatype::encode(args.datatype1), ""));
 562
 563   Colls::bcast(send_buffer(args.size * args.datatype1->size()), args.size, args.datatype1, args.root, MPI_COMM_WORLD);
 564
 565   TRACE_smpi_comm_out(my_proc_id);
 566 }
 567
 568 void ReduceAction::kernel(simgrid::xbt::ReplayAction&)
 569 {
 570   TRACE_smpi_comm_in(my_proc_id, "action_reduce",
 571       new simgrid::instr::CollTIData("reduce", MPI_COMM_WORLD->group()->actor(args.root)->get_pid(),
 572         args.comp_size, args.comm_size, -1,
 573         Datatype::encode(args.datatype1), ""));
 574
 575   Colls::reduce(send_buffer(args.comm_size * args.datatype1->size()),
 576       recv_buffer(args.comm_size * args.datatype1->size()), args.comm_size, args.datatype1, MPI_OP_NULL, args.root, MPI_COMM_WORLD);
 577   smpi_execute_flops(args.comp_size);
 578
 579   TRACE_smpi_comm_out(my_proc_id);
 580 }
 581
 582 void AllReduceAction::kernel(simgrid::xbt::ReplayAction&)
 583 {
 584   TRACE_smpi_comm_in(my_proc_id, "action_allreduce", new simgrid::instr::CollTIData("allreduce", -1, args.comp_size, args.comm_size, -1,
 585         Datatype::encode(args.datatype1), ""));
 586
 587   Colls::allreduce(send_buffer(args.comm_size * args.datatype1->size()),
 588       recv_buffer(args.comm_size * args.datatype1->size()), args.comm_size, args.datatype1, MPI_OP_NULL, MPI_COMM_WORLD);
 589   smpi_execute_flops(args.comp_size);
 590
 591   TRACE_smpi_comm_out(my_proc_id);
 592 }
 593
 594 void AllToAllAction::kernel(simgrid::xbt::ReplayAction&)
 595 {
 596   TRACE_smpi_comm_in(my_proc_id, "action_alltoall",
 597       new simgrid::instr::CollTIData("alltoall", -1, -1.0, args.send_size, args.recv_size,
 598         Datatype::encode(args.datatype1),
 599         Datatype::encode(args.datatype2)));
 600
 601   Colls::alltoall(send_buffer(args.send_size * args.comm_size * args.datatype1->size()), args.send_size,
 602       args.datatype1, recv_buffer(args.recv_size * args.comm_size * args.datatype2->size()),
 603       args.recv_size, args.datatype2, MPI_COMM_WORLD);
 604
 605   TRACE_smpi_comm_out(my_proc_id);
 606 }
 607
 608 void GatherAction::kernel(simgrid::xbt::ReplayAction&)
 609 {
 610   TRACE_smpi_comm_in(my_proc_id, name.c_str(), new simgrid::instr::CollTIData(name, (name == "gather") ? args.root : -1, -1.0, args.send_size, args.recv_size,
 611         Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
 612
 613   if (name == "gather") {
 614     int rank = MPI_COMM_WORLD->rank();
 615     Colls::gather(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
 616         (rank == args.root) ? recv_buffer(args.recv_size * args.comm_size * args.datatype2->size()) : nullptr, args.recv_size, args.datatype2, args.root, MPI_COMM_WORLD);
 617   }
 618   else
 619     Colls::allgather(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
 620         recv_buffer(args.recv_size * args.datatype2->size()), args.recv_size, args.datatype2, MPI_COMM_WORLD);
 621
 622   TRACE_smpi_comm_out(my_proc_id);
 623 }
 624
 625 void GatherVAction::kernel(simgrid::xbt::ReplayAction&)
 626 {
 627   int rank = MPI_COMM_WORLD->rank();
 628
 629   TRACE_smpi_comm_in(my_proc_id, name.c_str(), new simgrid::instr::VarCollTIData(
 630         name, (name == "gatherv") ? args.root : -1, args.send_size, nullptr, -1, args.recvcounts,
 631         Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
 632
 633   if (name == "gatherv") {
 634     Colls::gatherv(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
 635         (rank == args.root) ? recv_buffer(args.recv_size_sum * args.datatype2->size()) : nullptr,
 636         args.recvcounts->data(), args.disps.data(), args.datatype2, args.root, MPI_COMM_WORLD);
 637   }
 638   else {
 639     Colls::allgatherv(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
 640         recv_buffer(args.recv_size_sum * args.datatype2->size()), args.recvcounts->data(),
 641         args.disps.data(), args.datatype2, MPI_COMM_WORLD);
 642   }
 643
 644   TRACE_smpi_comm_out(my_proc_id);
 645 }
 646
 647 void ScatterAction::kernel(simgrid::xbt::ReplayAction&)
 648 {
 649   int rank = MPI_COMM_WORLD->rank();
 650   TRACE_smpi_comm_in(my_proc_id, "action_scatter", new simgrid::instr::CollTIData(name, args.root, -1.0, args.send_size, args.recv_size,
 651         Datatype::encode(args.datatype1),
 652         Datatype::encode(args.datatype2)));
 653
 654   Colls::scatter(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
 655       (rank == args.root) ? recv_buffer(args.recv_size * args.datatype2->size()) : nullptr, args.recv_size, args.datatype2, args.root, MPI_COMM_WORLD);
 656
 657   TRACE_smpi_comm_out(my_proc_id);
 658 }
 659
 660 void ScatterVAction::kernel(simgrid::xbt::ReplayAction&)
 661 {
 662   int rank = MPI_COMM_WORLD->rank();
 663   TRACE_smpi_comm_in(my_proc_id, "action_scatterv", new simgrid::instr::VarCollTIData(name, args.root, -1, args.sendcounts, args.recv_size,
 664         nullptr, Datatype::encode(args.datatype1),
 665         Datatype::encode(args.datatype2)));
 666
 667   Colls::scatterv((rank == args.root) ? send_buffer(args.send_size_sum * args.datatype1->size()) : nullptr,
 668       args.sendcounts->data(), args.disps.data(), args.datatype1,
 669       recv_buffer(args.recv_size * args.datatype2->size()), args.recv_size, args.datatype2, args.root,
 670       MPI_COMM_WORLD);
 671
 672   TRACE_smpi_comm_out(my_proc_id);
 673 }
 674
 675 void ReduceScatterAction::kernel(simgrid::xbt::ReplayAction&)
 676 {
 677   TRACE_smpi_comm_in(my_proc_id, "action_reducescatter",
 678       new simgrid::instr::VarCollTIData("reducescatter", -1, 0, nullptr, -1, args.recvcounts,
 679         std::to_string(args.comp_size), /* ugly hack to print comp_size */
 680         Datatype::encode(args.datatype1)));
 681
 682   Colls::reduce_scatter(send_buffer(args.recv_size_sum * args.datatype1->size()),
 683       recv_buffer(args.recv_size_sum * args.datatype1->size()), args.recvcounts->data(),
 684       args.datatype1, MPI_OP_NULL, MPI_COMM_WORLD);
 685
 686   smpi_execute_flops(args.comp_size);
 687   TRACE_smpi_comm_out(my_proc_id);
 688 }
 689
 690 void AllToAllVAction::kernel(simgrid::xbt::ReplayAction&)
 691 {
 692   TRACE_smpi_comm_in(my_proc_id, __func__,
 693       new simgrid::instr::VarCollTIData(
 694         "alltoallv", -1, args.send_size_sum, args.sendcounts, args.recv_size_sum, args.recvcounts,
 695         Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
 696
 697   Colls::alltoallv(send_buffer(args.send_buf_size * args.datatype1->size()), args.sendcounts->data(), args.senddisps.data(), args.datatype1,
 698       recv_buffer(args.recv_buf_size * args.datatype2->size()), args.recvcounts->data(), args.recvdisps.data(), args.datatype2, MPI_COMM_WORLD);
 699
 700   TRACE_smpi_comm_out(my_proc_id);
 701 }
 702 } // Replay Namespace
 703 }} // namespace simgrid::smpi
 704
 705 static std::unordered_map<aid_t, simgrid::smpi::replay::RequestStorage> storage;
 706 /** @brief Only initialize the replay, don't do it for real */
 707 void smpi_replay_init(const char* instance_id, int rank, double start_delay_flops)
 708 {
 709   if (not smpi_process()->initializing()){
 710     simgrid::s4u::Actor::self()->set_property("instance_id", instance_id);
 711     simgrid::s4u::Actor::self()->set_property("rank", std::to_string(rank));
 712     simgrid::smpi::ActorExt::init();
 713   }
 714   smpi_process()->mark_as_initialized();
 715   smpi_process()->set_replaying(true);
 716
 717   int my_proc_id = simgrid::s4u::this_actor::get_pid();
 718
 719   TRACE_smpi_init(my_proc_id);
 720   TRACE_smpi_computing_init(my_proc_id);
 721   TRACE_smpi_comm_in(my_proc_id, "smpi_replay_run_init", new simgrid::instr::NoOpTIData("init"));
 722   TRACE_smpi_comm_out(my_proc_id);
 723   xbt_replay_action_register("init", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::InitAction().execute(action); });
 724   xbt_replay_action_register("finalize", [](simgrid::xbt::ReplayAction&) { /* nothing to do */ });
 725   xbt_replay_action_register("comm_size", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
 726   xbt_replay_action_register("comm_split",[](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
 727   xbt_replay_action_register("comm_dup",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
 728   xbt_replay_action_register("send",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::SendAction("send", storage[simgrid::s4u::this_actor::get_pid()]).execute(action); });
 729   xbt_replay_action_register("isend", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::SendAction("isend", storage[simgrid::s4u::this_actor::get_pid()]).execute(action); });
 730   xbt_replay_action_register("recv",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::RecvAction("recv", storage[simgrid::s4u::this_actor::get_pid()]).execute(action); });
 731   xbt_replay_action_register("irecv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::RecvAction("irecv", storage[simgrid::s4u::this_actor::get_pid()]).execute(action); });
 732   xbt_replay_action_register("test",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::TestAction(storage[simgrid::s4u::this_actor::get_pid()]).execute(action); });
 733   xbt_replay_action_register("wait",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::WaitAction(storage[simgrid::s4u::this_actor::get_pid()]).execute(action); });
 734   xbt_replay_action_register("waitall", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::WaitAllAction(storage[simgrid::s4u::this_actor::get_pid()]).execute(action); });
 735   xbt_replay_action_register("barrier", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::BarrierAction().execute(action); });
 736   xbt_replay_action_register("bcast",   [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::BcastAction().execute(action); });
 737   xbt_replay_action_register("reduce",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ReduceAction().execute(action); });
 738   xbt_replay_action_register("allreduce", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllReduceAction().execute(action); });
 739   xbt_replay_action_register("alltoall", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllToAllAction().execute(action); });
 740   xbt_replay_action_register("alltoallv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllToAllVAction().execute(action); });
 741   xbt_replay_action_register("gather",   [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherAction("gather").execute(action); });
 742   xbt_replay_action_register("scatter",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ScatterAction().execute(action); });
 743   xbt_replay_action_register("gatherv",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherVAction("gatherv").execute(action); });
 744   xbt_replay_action_register("scatterv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ScatterVAction().execute(action); });
 745   xbt_replay_action_register("allgather", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherAction("allgather").execute(action); });
 746   xbt_replay_action_register("allgatherv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherVAction("allgatherv").execute(action); });
 747   xbt_replay_action_register("reducescatter", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ReduceScatterAction().execute(action); });
 748   xbt_replay_action_register("compute", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ComputeAction().execute(action); });
 749
 750   //if we have a delayed start, sleep here.
 751   if (start_delay_flops > 0) {
 752     XBT_VERB("Delayed start for instance - Sleeping for %f flops ", start_delay_flops);
 753     smpi_execute_flops(start_delay_flops);
 754   } else {
 755     // Wait for the other actors to initialize also
 756     simgrid::s4u::this_actor::yield();
 757   }
 758 }
 759
 760 /** @brief actually run the replay after initialization */
 761 void smpi_replay_main(int rank, const char* trace_filename)
 762 {
 763   static int active_processes = 0;
 764   active_processes++;
 765   storage[simgrid::s4u::this_actor::get_pid()] = simgrid::smpi::replay::RequestStorage();
 766   std::string rank_string                      = std::to_string(rank);
 767   simgrid::xbt::replay_runner(rank_string.c_str(), trace_filename);
 768
 769   /* and now, finalize everything */
 770   /* One active process will stop. Decrease the counter*/
 771   unsigned int count_requests = storage[simgrid::s4u::this_actor::get_pid()].size();
 772   XBT_DEBUG("There are %ud elements in reqq[*]", count_requests);
 773   if (count_requests > 0) {
 774     MPI_Request requests[count_requests];
 775     MPI_Status status[count_requests];
 776     unsigned int i=0;
 777
 778     for (auto const& pair : storage[simgrid::s4u::this_actor::get_pid()].get_store()) {
 779       requests[i] = pair.second;
 780       i++;
 781     }
 782     simgrid::smpi::Request::waitall(count_requests, requests, status);
 783   }
 784   active_processes--;
 785
 786   if(active_processes==0){
 787     /* Last process alive speaking: end the simulated timer */
 788     XBT_INFO("Simulation time %f", smpi_process()->simulated_elapsed());
 789     smpi_free_replay_tmp_buffers();
 790   }
 791
 792   TRACE_smpi_comm_in(simgrid::s4u::this_actor::get_pid(), "smpi_replay_run_finalize",
 793                      new simgrid::instr::NoOpTIData("finalize"));
 794
 795   smpi_process()->finalize();
 796
 797   TRACE_smpi_comm_out(simgrid::s4u::this_actor::get_pid());
 798   TRACE_smpi_finalize(simgrid::s4u::this_actor::get_pid());
 799 }
 800
 801 /** @brief chain a replay initialization and a replay start */
 802 void smpi_replay_run(const char* instance_id, int rank, double start_delay_flops, const char* trace_filename)
 803 {
 804   smpi_replay_init(instance_id, rank, start_delay_flops);
 805   smpi_replay_main(rank, trace_filename);
 806 }