Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
72b7be079c53bb38fe8c3b9f370090b4a644cdb2
[simgrid.git] / src / smpi / internals / smpi_replay.cpp
1 /* Copyright (c) 2009-2018. 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 "private.hpp"
7 #include "smpi_coll.hpp"
8 #include "smpi_comm.hpp"
9 #include "smpi_datatype.hpp"
10 #include "smpi_group.hpp"
11 #include "smpi_process.hpp"
12 #include "smpi_request.hpp"
13 #include "xbt/replay.hpp"
14
15 #include <boost/algorithm/string/join.hpp>
16 #include <memory>
17 #include <numeric>
18 #include <unordered_map>
19 #include <sstream>
20 #include <vector>
21
22 using simgrid::s4u::Actor;
23
24 #include <tuple>
25 // From https://stackoverflow.com/questions/7110301/generic-hash-for-tuples-in-unordered-map-unordered-set
26 // This is all just to make std::unordered_map work with std::tuple. If we need this in other places,
27 // this could go into a header file.
28 namespace hash_tuple{
29     template <typename TT>
30     struct hash
31     {
32         size_t
33         operator()(TT const& tt) const
34         {
35             return std::hash<TT>()(tt);
36         }
37     };
38
39     template <class T>
40     inline void hash_combine(std::size_t& seed, T const& v)
41     {
42         seed ^= hash_tuple::hash<T>()(v) + 0x9e3779b9 + (seed<<6) + (seed>>2);
43     }
44
45     // Recursive template code derived from Matthieu M.
46     template <class Tuple, size_t Index = std::tuple_size<Tuple>::value - 1>
47     struct HashValueImpl
48     {
49       static void apply(size_t& seed, Tuple const& tuple)
50       {
51         HashValueImpl<Tuple, Index-1>::apply(seed, tuple);
52         hash_combine(seed, std::get<Index>(tuple));
53       }
54     };
55
56     template <class Tuple>
57     struct HashValueImpl<Tuple,0>
58     {
59       static void apply(size_t& seed, Tuple const& tuple)
60       {
61         hash_combine(seed, std::get<0>(tuple));
62       }
63     };
64
65     template <typename ... TT>
66     struct hash<std::tuple<TT...>>
67     {
68         size_t
69         operator()(std::tuple<TT...> const& tt) const
70         {
71             size_t seed = 0;
72             HashValueImpl<std::tuple<TT...> >::apply(seed, tt);
73             return seed;
74         }
75     };
76 }
77
78 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_replay,smpi,"Trace Replay with SMPI");
79
80 static std::unordered_map<int, std::vector<MPI_Request>*> reqq;
81
82 static MPI_Datatype MPI_DEFAULT_TYPE;
83
84 #define CHECK_ACTION_PARAMS(action, mandatory, optional)                                                               \
85   {                                                                                                                    \
86     if (action.size() < static_cast<unsigned long>(mandatory + 2)) {                                                   \
87       std::stringstream ss;                                                                                            \
88       for (const auto& elem : action) {                                                                                \
89         ss << elem << " ";                                                                                             \
90       }                                                                                                                \
91       THROWF(arg_error, 0, "%s replay failed.\n"                                                                       \
92                            "%zu items were given on the line. First two should be process_id and action.  "            \
93                            "This action needs after them %lu mandatory arguments, and accepts %lu optional ones. \n"   \
94                            "The full line that was given is:\n   %s\n"                                                 \
95                            "Please contact the Simgrid team if support is needed",                                     \
96              __func__, action.size(), static_cast<unsigned long>(mandatory), static_cast<unsigned long>(optional),     \
97              ss.str().c_str());                                                                                        \
98     }                                                                                                                  \
99   }
100
101 static void log_timed_action(simgrid::xbt::ReplayAction& action, double clock)
102 {
103   if (XBT_LOG_ISENABLED(smpi_replay, xbt_log_priority_verbose)){
104     std::string s = boost::algorithm::join(action, " ");
105     XBT_VERB("%s %f", s.c_str(), smpi_process()->simulated_elapsed() - clock);
106   }
107 }
108
109 static std::vector<MPI_Request>* get_reqq_self()
110 {
111   return reqq.at(simgrid::s4u::this_actor::get_pid());
112 }
113
114 static void set_reqq_self(std::vector<MPI_Request> *mpi_request)
115 {
116   reqq.insert({simgrid::s4u::this_actor::get_pid(), mpi_request});
117 }
118
119 /* Helper function */
120 static double parse_double(std::string string)
121 {
122   return xbt_str_parse_double(string.c_str(), "%s is not a double");
123 }
124
125 namespace simgrid {
126 namespace smpi {
127
128 namespace replay {
129
130 class RequestStorage {
131 private:
132     req_storage_t store;
133
134 public:
135     RequestStorage() {}
136     int size()
137     {
138       return store.size();
139     }
140
141     req_storage_t& get_store()
142     {
143       return store;
144     }
145
146     void get_requests(std::vector<MPI_Request>& vec)
147     {
148       for (auto& pair : store) {
149         auto& req = pair.second;
150         auto my_proc_id = simgrid::s4u::this_actor::getPid();
151         if (req != MPI_REQUEST_NULL && (req->src() == my_proc_id || req->dst() == my_proc_id)) {
152           vec.push_back(pair.second);
153           pair.second->print_request("MM");
154         }
155       }
156     }
157
158     MPI_Request find(int src, int dst, int tag)
159     {
160       req_storage_t::iterator it = store.find(req_key_t(src, dst, tag));
161       return (it == store.end()) ? MPI_REQUEST_NULL : it->second;
162     }
163
164     void remove(MPI_Request req)
165     {
166       if (req == MPI_REQUEST_NULL) return;
167
168       store.erase(req_key_t(req->src()-1, req->dst()-1, req->tag()));
169     }
170
171     void add(MPI_Request req)
172     {
173       if (req != MPI_REQUEST_NULL) // Can and does happen in the case of TestAction
174         store.insert({req_key_t(req->src()-1, req->dst()-1, req->tag()), req});
175     }
176
177     /* Sometimes we need to re-insert MPI_REQUEST_NULL but we still need src,dst and tag */
178     void addNullRequest(int src, int dst, int tag)
179     {
180       store.insert({req_key_t(src, dst, tag), MPI_REQUEST_NULL});
181     }
182 };
183
184 class ActionArgParser {
185 public:
186   virtual ~ActionArgParser() = default;
187   virtual void parse(simgrid::xbt::ReplayAction& action, std::string name) { CHECK_ACTION_PARAMS(action, 0, 0) }
188 };
189
190 class WaitTestParser : public ActionArgParser {
191 public:
192   int src;
193   int dst;
194   int tag;
195
196   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
197   {
198     CHECK_ACTION_PARAMS(action, 3, 0)
199     src = std::stoi(action[2]);
200     dst = std::stoi(action[3]);
201     tag = std::stoi(action[4]);
202   }
203 };
204
205 class SendRecvParser : public ActionArgParser {
206 public:
207   /* communication partner; if we send, this is the receiver and vice versa */
208   int partner;
209   double size;
210   int tag;
211   MPI_Datatype datatype1 = MPI_DEFAULT_TYPE;
212
213   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
214   {
215     CHECK_ACTION_PARAMS(action, 3, 1)
216     partner = std::stoi(action[2]);
217     tag     = std::stoi(action[3]);
218     size    = parse_double(action[4]);
219     if (action.size() > 5)
220       datatype1 = simgrid::smpi::Datatype::decode(action[5]);
221   }
222 };
223
224 class ComputeParser : public ActionArgParser {
225 public:
226   /* communication partner; if we send, this is the receiver and vice versa */
227   double flops;
228
229   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
230   {
231     CHECK_ACTION_PARAMS(action, 1, 0)
232     flops = parse_double(action[2]);
233   }
234 };
235
236 class CollCommParser : public ActionArgParser {
237 public:
238   double size;
239   double comm_size;
240   double comp_size;
241   int send_size;
242   int recv_size;
243   int root = 0;
244   MPI_Datatype datatype1 = MPI_DEFAULT_TYPE;
245   MPI_Datatype datatype2 = MPI_DEFAULT_TYPE;
246 };
247
248 class BcastArgParser : public CollCommParser {
249 public:
250   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
251   {
252     CHECK_ACTION_PARAMS(action, 1, 2)
253     size = parse_double(action[2]);
254     root = (action.size() > 3) ? std::stoi(action[3]) : 0;
255     if (action.size() > 4)
256       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
257   }
258 };
259
260 class ReduceArgParser : public CollCommParser {
261 public:
262   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
263   {
264     CHECK_ACTION_PARAMS(action, 2, 2)
265     comm_size = parse_double(action[2]);
266     comp_size = parse_double(action[3]);
267     root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
268     if (action.size() > 5)
269       datatype1 = simgrid::smpi::Datatype::decode(action[5]);
270   }
271 };
272
273 class AllReduceArgParser : public CollCommParser {
274 public:
275   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
276   {
277     CHECK_ACTION_PARAMS(action, 2, 1)
278     comm_size = parse_double(action[2]);
279     comp_size = parse_double(action[3]);
280     if (action.size() > 4)
281       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
282   }
283 };
284
285 class AllToAllArgParser : public CollCommParser {
286 public:
287   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
288   {
289     CHECK_ACTION_PARAMS(action, 2, 1)
290     comm_size = MPI_COMM_WORLD->size();
291     send_size = parse_double(action[2]);
292     recv_size = parse_double(action[3]);
293
294     if (action.size() > 4)
295       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
296     if (action.size() > 5)
297       datatype2 = simgrid::smpi::Datatype::decode(action[5]);
298   }
299 };
300
301 class GatherArgParser : public CollCommParser {
302 public:
303   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
304   {
305     /* The structure of the gather action for the rank 0 (total 4 processes) is the following:
306           0 gather 68 68 0 0 0
307         where:
308           1) 68 is the sendcounts
309           2) 68 is the recvcounts
310           3) 0 is the root node
311           4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
312           5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
313     */
314     CHECK_ACTION_PARAMS(action, 2, 3)
315     comm_size = MPI_COMM_WORLD->size();
316     send_size = parse_double(action[2]);
317     recv_size = parse_double(action[3]);
318
319     if (name == "gather") {
320       root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
321       if (action.size() > 5)
322         datatype1 = simgrid::smpi::Datatype::decode(action[5]);
323       if (action.size() > 6)
324         datatype2 = simgrid::smpi::Datatype::decode(action[6]);
325     }
326     else {
327       if (action.size() > 4)
328         datatype1 = simgrid::smpi::Datatype::decode(action[4]);
329       if (action.size() > 5)
330         datatype2 = simgrid::smpi::Datatype::decode(action[5]);
331     }
332   }
333 };
334
335 class GatherVArgParser : public CollCommParser {
336 public:
337   int recv_size_sum;
338   std::shared_ptr<std::vector<int>> recvcounts;
339   std::vector<int> disps;
340   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
341   {
342     /* The structure of the gatherv action for the rank 0 (total 4 processes) is the following:
343          0 gather 68 68 10 10 10 0 0 0
344        where:
345          1) 68 is the sendcount
346          2) 68 10 10 10 is the recvcounts
347          3) 0 is the root node
348          4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
349          5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
350     */
351     comm_size = MPI_COMM_WORLD->size();
352     CHECK_ACTION_PARAMS(action, comm_size+1, 2)
353     send_size = parse_double(action[2]);
354     disps     = std::vector<int>(comm_size, 0);
355     recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
356
357     if (name == "gatherV") {
358       root = (action.size() > 3 + comm_size) ? std::stoi(action[3 + comm_size]) : 0;
359       if (action.size() > 4 + comm_size)
360         datatype1 = simgrid::smpi::Datatype::decode(action[4 + comm_size]);
361       if (action.size() > 5 + comm_size)
362         datatype2 = simgrid::smpi::Datatype::decode(action[5 + comm_size]);
363     }
364     else {
365       int datatype_index = 0;
366       int disp_index     = 0;
367       /* The 3 comes from "0 gather <sendcount>", which must always be present.
368        * The + comm_size is the recvcounts array, which must also be present
369        */
370       if (action.size() > 3 + comm_size + comm_size) { /* datatype + disp are specified */
371         datatype_index = 3 + comm_size;
372         disp_index     = datatype_index + 1;
373         datatype1      = simgrid::smpi::Datatype::decode(action[datatype_index]);
374         datatype2      = simgrid::smpi::Datatype::decode(action[datatype_index]);
375       } else if (action.size() > 3 + comm_size + 2) { /* disps specified; datatype is not specified; use the default one */
376         disp_index     = 3 + comm_size;
377       } else if (action.size() > 3 + comm_size)  { /* only datatype, no disp specified */
378         datatype_index = 3 + comm_size;
379         datatype1      = simgrid::smpi::Datatype::decode(action[datatype_index]);
380         datatype2      = simgrid::smpi::Datatype::decode(action[datatype_index]);
381       }
382
383       if (disp_index != 0) {
384         for (unsigned int i = 0; i < comm_size; i++)
385           disps[i]          = std::stoi(action[disp_index + i]);
386       }
387     }
388
389     for (unsigned int i = 0; i < comm_size; i++) {
390       (*recvcounts)[i] = std::stoi(action[i + 3]);
391     }
392     recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
393   }
394 };
395
396 class ScatterArgParser : public CollCommParser {
397 public:
398   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
399   {
400     /* The structure of the scatter action for the rank 0 (total 4 processes) is the following:
401           0 gather 68 68 0 0 0
402         where:
403           1) 68 is the sendcounts
404           2) 68 is the recvcounts
405           3) 0 is the root node
406           4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
407           5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
408     */
409     CHECK_ACTION_PARAMS(action, 2, 3)
410     comm_size   = MPI_COMM_WORLD->size();
411     send_size   = parse_double(action[2]);
412     recv_size   = parse_double(action[3]);
413     root   = (action.size() > 4) ? std::stoi(action[4]) : 0;
414     if (action.size() > 5)
415       datatype1 = simgrid::smpi::Datatype::decode(action[5]);
416     if (action.size() > 6)
417       datatype2 = simgrid::smpi::Datatype::decode(action[6]);
418   }
419 };
420
421 class ScatterVArgParser : public CollCommParser {
422 public:
423   int recv_size_sum;
424   int send_size_sum;
425   std::shared_ptr<std::vector<int>> sendcounts;
426   std::vector<int> disps;
427   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
428   {
429     /* The structure of the scatterv action for the rank 0 (total 4 processes) is the following:
430        0 gather 68 10 10 10 68 0 0 0
431         where:
432         1) 68 10 10 10 is the sendcounts
433         2) 68 is the recvcount
434         3) 0 is the root node
435         4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
436         5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
437     */
438     CHECK_ACTION_PARAMS(action, comm_size + 1, 2)
439     recv_size  = parse_double(action[2 + comm_size]);
440     disps      = std::vector<int>(comm_size, 0);
441     sendcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
442
443     if (action.size() > 5 + comm_size)
444       datatype1 = simgrid::smpi::Datatype::decode(action[4 + comm_size]);
445     if (action.size() > 5 + comm_size)
446       datatype2 = simgrid::smpi::Datatype::decode(action[5]);
447
448     for (unsigned int i = 0; i < comm_size; i++) {
449       (*sendcounts)[i] = std::stoi(action[i + 2]);
450     }
451     send_size_sum = std::accumulate(sendcounts->begin(), sendcounts->end(), 0);
452     root = (action.size() > 3 + comm_size) ? std::stoi(action[3 + comm_size]) : 0;
453   }
454 };
455
456 class ReduceScatterArgParser : public CollCommParser {
457 public:
458   int recv_size_sum;
459   std::shared_ptr<std::vector<int>> recvcounts;
460   std::vector<int> disps;
461   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
462   {
463     /* The structure of the reducescatter action for the rank 0 (total 4 processes) is the following:
464          0 reduceScatter 275427 275427 275427 204020 11346849 0
465        where:
466          1) The first four values after the name of the action declare the recvcounts array
467          2) The value 11346849 is the amount of instructions
468          3) The last value corresponds to the datatype, see simgrid::smpi::Datatype::decode().
469     */
470     comm_size = MPI_COMM_WORLD->size();
471     CHECK_ACTION_PARAMS(action, comm_size+1, 1)
472     comp_size = parse_double(action[2+comm_size]);
473     recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
474     if (action.size() > 3 + comm_size)
475       datatype1 = simgrid::smpi::Datatype::decode(action[3 + comm_size]);
476
477     for (unsigned int i = 0; i < comm_size; i++) {
478       recvcounts->push_back(std::stoi(action[i + 2]));
479     }
480     recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
481   }
482 };
483
484 class AllToAllVArgParser : public CollCommParser {
485 public:
486   int recv_size_sum;
487   int send_size_sum;
488   std::shared_ptr<std::vector<int>> recvcounts;
489   std::shared_ptr<std::vector<int>> sendcounts;
490   std::vector<int> senddisps;
491   std::vector<int> recvdisps;
492   int send_buf_size;
493   int recv_buf_size;
494   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
495   {
496     /* The structure of the allToAllV action for the rank 0 (total 4 processes) is the following:
497           0 allToAllV 100 1 7 10 12 100 1 70 10 5
498        where:
499         1) 100 is the size of the send buffer *sizeof(int),
500         2) 1 7 10 12 is the sendcounts array
501         3) 100*sizeof(int) is the size of the receiver buffer
502         4)  1 70 10 5 is the recvcounts array
503     */
504     comm_size = MPI_COMM_WORLD->size();
505     CHECK_ACTION_PARAMS(action, 2*comm_size+2, 2)
506     sendcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
507     recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
508     senddisps  = std::vector<int>(comm_size, 0);
509     recvdisps  = std::vector<int>(comm_size, 0);
510
511     if (action.size() > 5 + 2 * comm_size)
512       datatype1 = simgrid::smpi::Datatype::decode(action[4 + 2 * comm_size]);
513     if (action.size() > 5 + 2 * comm_size)
514       datatype2 = simgrid::smpi::Datatype::decode(action[5 + 2 * comm_size]);
515
516     send_buf_size=parse_double(action[2]);
517     recv_buf_size=parse_double(action[3+comm_size]);
518     for (unsigned int i = 0; i < comm_size; i++) {
519       (*sendcounts)[i] = std::stoi(action[3 + i]);
520       (*recvcounts)[i] = std::stoi(action[4 + comm_size + i]);
521     }
522     send_size_sum = std::accumulate(sendcounts->begin(), sendcounts->end(), 0);
523     recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
524   }
525 };
526
527 template <class T> class ReplayAction {
528 protected:
529   const std::string name;
530   const int my_proc_id;
531   T args;
532
533 public:
534   explicit ReplayAction(std::string name) : name(name), my_proc_id(simgrid::s4u::this_actor::get_pid()) {}
535   virtual ~ReplayAction() = default;
536
537   virtual void execute(simgrid::xbt::ReplayAction& action)
538   {
539     // Needs to be re-initialized for every action, hence here
540     double start_time = smpi_process()->simulated_elapsed();
541     args.parse(action, name);
542     kernel(action);
543     if (name != "Init")
544       log_timed_action(action, start_time);
545   }
546
547   virtual void kernel(simgrid::xbt::ReplayAction& action) = 0;
548
549   void* send_buffer(int size)
550   {
551     return smpi_get_tmp_sendbuffer(size);
552   }
553
554   void* recv_buffer(int size)
555   {
556     return smpi_get_tmp_recvbuffer(size);
557   }
558 };
559
560 class WaitAction : public ReplayAction<WaitTestParser> {
561 public:
562   WaitAction() : ReplayAction("Wait") {}
563   void kernel(simgrid::xbt::ReplayAction& action) override
564   {
565     std::string s = boost::algorithm::join(action, " ");
566     xbt_assert(get_reqq_self()->size(), "action wait not preceded by any irecv or isend: %s", s.c_str());
567     MPI_Request request = get_reqq_self()->back();
568     get_reqq_self()->pop_back();
569
570     if (request == MPI_REQUEST_NULL) {
571       /* Assume that the trace is well formed, meaning the comm might have been caught by a MPI_test. Then just
572        * return.*/
573       return;
574     }
575
576     int rank = request->comm() != MPI_COMM_NULL ? request->comm()->rank() : -1;
577
578     // Must be taken before Request::wait() since the request may be set to
579     // MPI_REQUEST_NULL by Request::wait!
580     bool is_wait_for_receive = (request->flags() & RECV);
581     // TODO: Here we take the rank while we normally take the process id (look for my_proc_id)
582     TRACE_smpi_comm_in(rank, __func__, new simgrid::instr::NoOpTIData("wait"));
583
584     MPI_Status status;
585     Request::wait(&request, &status);
586
587     TRACE_smpi_comm_out(rank);
588     if (is_wait_for_receive)
589       TRACE_smpi_recv(args.src, args.dst, args.tag);
590   }
591 };
592
593 class SendAction : public ReplayAction<SendRecvParser> {
594 public:
595   SendAction() = delete;
596   explicit SendAction(std::string name) : ReplayAction(name) {}
597   void kernel(simgrid::xbt::ReplayAction& action) override
598   {
599     int dst_traced = MPI_COMM_WORLD->group()->actor(args.partner)->get_pid();
600
601     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData(name, args.partner, args.size,
602                                                                              args.tag, Datatype::encode(args.datatype1)));
603     if (not TRACE_smpi_view_internals())
604       TRACE_smpi_send(my_proc_id, my_proc_id, dst_traced, args.tag, args.size * args.datatype1->size());
605
606     if (name == "send") {
607       Request::send(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
608     } else if (name == "Isend") {
609       MPI_Request request = Request::isend(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
610       get_reqq_self()->push_back(request);
611     } else {
612       xbt_die("Don't know this action, %s", name.c_str());
613     }
614
615     TRACE_smpi_comm_out(my_proc_id);
616   }
617 };
618
619 class RecvAction : public ReplayAction<SendRecvParser> {
620 public:
621   RecvAction() = delete;
622   explicit RecvAction(std::string name) : ReplayAction(name) {}
623   void kernel(simgrid::xbt::ReplayAction& action) override
624   {
625     int src_traced = MPI_COMM_WORLD->group()->actor(args.partner)->get_pid();
626
627     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData(name, args.partner, args.size,
628                                                                              args.tag, Datatype::encode(args.datatype1)));
629
630     MPI_Status status;
631     // unknown size from the receiver point of view
632     if (args.size <= 0.0) {
633       Request::probe(args.partner, args.tag, MPI_COMM_WORLD, &status);
634       args.size = status.count;
635     }
636
637     if (name == "recv") {
638       Request::recv(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD, &status);
639     } else if (name == "Irecv") {
640       MPI_Request request = Request::irecv(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
641       get_reqq_self()->push_back(request);
642     }
643
644     TRACE_smpi_comm_out(my_proc_id);
645     // TODO: Check why this was only activated in the "recv" case and not in the "Irecv" case
646     if (name == "recv" && not TRACE_smpi_view_internals()) {
647       TRACE_smpi_recv(src_traced, my_proc_id, args.tag);
648     }
649   }
650 };
651
652 class ComputeAction : public ReplayAction<ComputeParser> {
653 public:
654   ComputeAction() : ReplayAction("compute") {}
655   void kernel(simgrid::xbt::ReplayAction& action) override
656   {
657     TRACE_smpi_computing_in(my_proc_id, args.flops);
658     smpi_execute_flops(args.flops);
659     TRACE_smpi_computing_out(my_proc_id);
660   }
661 };
662
663 class TestAction : public ReplayAction<WaitTestParser> {
664 public:
665   TestAction() : ReplayAction("Test") {}
666   void kernel(simgrid::xbt::ReplayAction& action) override
667   {
668     MPI_Request request = get_reqq_self()->back();
669     get_reqq_self()->pop_back();
670     // if request is null here, this may mean that a previous test has succeeded
671     // Different times in traced application and replayed version may lead to this
672     // In this case, ignore the extra calls.
673     if (request != MPI_REQUEST_NULL) {
674       TRACE_smpi_testing_in(my_proc_id);
675
676       MPI_Status status;
677       int flag = Request::test(&request, &status);
678
679       XBT_DEBUG("MPI_Test result: %d", flag);
680       /* push back request in vector to be caught by a subsequent wait. if the test did succeed, the request is now
681        * nullptr.*/
682       get_reqq_self()->push_back(request);
683
684       TRACE_smpi_testing_out(my_proc_id);
685     }
686   }
687 };
688
689 class InitAction : public ReplayAction<ActionArgParser> {
690 public:
691   InitAction() : ReplayAction("Init") {}
692   void kernel(simgrid::xbt::ReplayAction& action) override
693   {
694     CHECK_ACTION_PARAMS(action, 0, 1)
695     MPI_DEFAULT_TYPE = (action.size() > 2) ? MPI_DOUBLE // default MPE datatype
696                                            : MPI_BYTE;  // default TAU datatype
697
698     /* start a simulated timer */
699     smpi_process()->simulated_start();
700     set_reqq_self(new std::vector<MPI_Request>);
701   }
702 };
703
704 class CommunicatorAction : public ReplayAction<ActionArgParser> {
705 public:
706   CommunicatorAction() : ReplayAction("Comm") {}
707   void kernel(simgrid::xbt::ReplayAction& action) override { /* nothing to do */}
708 };
709
710 class WaitAllAction : public ReplayAction<ActionArgParser> {
711 public:
712   WaitAllAction() : ReplayAction("waitAll") {}
713   void kernel(simgrid::xbt::ReplayAction& action) override
714   {
715     const unsigned int count_requests = get_reqq_self()->size();
716
717     if (count_requests > 0) {
718       TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData("waitAll", -1, count_requests, ""));
719       std::vector<std::pair</*sender*/int,/*recv*/int>> sender_receiver;
720       for (const auto& req : (*get_reqq_self())) {
721         if (req && (req->flags() & RECV)) {
722           sender_receiver.push_back({req->src(), req->dst()});
723         }
724       }
725       MPI_Status status[count_requests];
726       Request::waitall(count_requests, &(*get_reqq_self())[0], status);
727
728       for (auto& pair : sender_receiver) {
729         TRACE_smpi_recv(pair.first, pair.second, 0);
730       }
731       TRACE_smpi_comm_out(my_proc_id);
732     }
733   }
734 };
735
736 class BarrierAction : public ReplayAction<ActionArgParser> {
737 public:
738   BarrierAction() : ReplayAction("barrier") {}
739   void kernel(simgrid::xbt::ReplayAction& action) override
740   {
741     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::NoOpTIData("barrier"));
742     Colls::barrier(MPI_COMM_WORLD);
743     TRACE_smpi_comm_out(my_proc_id);
744   }
745 };
746
747 class BcastAction : public ReplayAction<BcastArgParser> {
748 public:
749   BcastAction() : ReplayAction("bcast") {}
750   void kernel(simgrid::xbt::ReplayAction& action) override
751   {
752     TRACE_smpi_comm_in(my_proc_id, "action_bcast",
753                        new simgrid::instr::CollTIData("bcast", MPI_COMM_WORLD->group()->actor(args.root)->get_pid(),
754                                                       -1.0, args.size, -1, Datatype::encode(args.datatype1), ""));
755
756     Colls::bcast(send_buffer(args.size * args.datatype1->size()), args.size, args.datatype1, args.root, MPI_COMM_WORLD);
757
758     TRACE_smpi_comm_out(my_proc_id);
759   }
760 };
761
762 class ReduceAction : public ReplayAction<ReduceArgParser> {
763 public:
764   ReduceAction() : ReplayAction("reduce") {}
765   void kernel(simgrid::xbt::ReplayAction& action) override
766   {
767     TRACE_smpi_comm_in(my_proc_id, "action_reduce",
768                        new simgrid::instr::CollTIData("reduce", MPI_COMM_WORLD->group()->actor(args.root)->get_pid(),
769                                                       args.comp_size, args.comm_size, -1,
770                                                       Datatype::encode(args.datatype1), ""));
771
772     Colls::reduce(send_buffer(args.comm_size * args.datatype1->size()),
773         recv_buffer(args.comm_size * args.datatype1->size()), args.comm_size, args.datatype1, MPI_OP_NULL, args.root, MPI_COMM_WORLD);
774     smpi_execute_flops(args.comp_size);
775
776     TRACE_smpi_comm_out(my_proc_id);
777   }
778 };
779
780 class AllReduceAction : public ReplayAction<AllReduceArgParser> {
781 public:
782   AllReduceAction() : ReplayAction("allReduce") {}
783   void kernel(simgrid::xbt::ReplayAction& action) override
784   {
785     TRACE_smpi_comm_in(my_proc_id, "action_allReduce", new simgrid::instr::CollTIData("allReduce", -1, args.comp_size, args.comm_size, -1,
786                                                                                 Datatype::encode(args.datatype1), ""));
787
788     Colls::allreduce(send_buffer(args.comm_size * args.datatype1->size()),
789         recv_buffer(args.comm_size * args.datatype1->size()), args.comm_size, args.datatype1, MPI_OP_NULL, MPI_COMM_WORLD);
790     smpi_execute_flops(args.comp_size);
791
792     TRACE_smpi_comm_out(my_proc_id);
793   }
794 };
795
796 class AllToAllAction : public ReplayAction<AllToAllArgParser> {
797 public:
798   AllToAllAction() : ReplayAction("allToAll") {}
799   void kernel(simgrid::xbt::ReplayAction& action) override
800   {
801     TRACE_smpi_comm_in(my_proc_id, "action_allToAll",
802                      new simgrid::instr::CollTIData("allToAll", -1, -1.0, args.send_size, args.recv_size,
803                                                     Datatype::encode(args.datatype1),
804                                                     Datatype::encode(args.datatype2)));
805
806     Colls::alltoall(send_buffer(args.send_size * args.comm_size * args.datatype1->size()), args.send_size,
807                     args.datatype1, recv_buffer(args.recv_size * args.comm_size * args.datatype2->size()),
808                     args.recv_size, args.datatype2, MPI_COMM_WORLD);
809
810     TRACE_smpi_comm_out(my_proc_id);
811   }
812 };
813
814 class GatherAction : public ReplayAction<GatherArgParser> {
815 public:
816   explicit GatherAction(std::string name) : ReplayAction(name) {}
817   void kernel(simgrid::xbt::ReplayAction& action) override
818   {
819     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,
820                                                                           Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
821
822     if (name == "gather") {
823       int rank = MPI_COMM_WORLD->rank();
824       Colls::gather(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
825                  (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);
826     }
827     else
828       Colls::allgather(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
829                        recv_buffer(args.recv_size * args.datatype2->size()), args.recv_size, args.datatype2, MPI_COMM_WORLD);
830
831     TRACE_smpi_comm_out(my_proc_id);
832   }
833 };
834
835 class GatherVAction : public ReplayAction<GatherVArgParser> {
836 public:
837   explicit GatherVAction(std::string name) : ReplayAction(name) {}
838   void kernel(simgrid::xbt::ReplayAction& action) override
839   {
840     int rank = MPI_COMM_WORLD->rank();
841
842     TRACE_smpi_comm_in(my_proc_id, name.c_str(), new simgrid::instr::VarCollTIData(
843                                                name, (name == "gatherV") ? args.root : -1, args.send_size, nullptr, -1, args.recvcounts,
844                                                Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
845
846     if (name == "gatherV") {
847       Colls::gatherv(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
848                      (rank == args.root) ? recv_buffer(args.recv_size_sum * args.datatype2->size()) : nullptr,
849                      args.recvcounts->data(), args.disps.data(), args.datatype2, args.root, MPI_COMM_WORLD);
850     }
851     else {
852       Colls::allgatherv(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
853                         recv_buffer(args.recv_size_sum * args.datatype2->size()), args.recvcounts->data(),
854                         args.disps.data(), args.datatype2, MPI_COMM_WORLD);
855     }
856
857     TRACE_smpi_comm_out(my_proc_id);
858   }
859 };
860
861 class ScatterAction : public ReplayAction<ScatterArgParser> {
862 public:
863   ScatterAction() : ReplayAction("scatter") {}
864   void kernel(simgrid::xbt::ReplayAction& action) override
865   {
866     int rank = MPI_COMM_WORLD->rank();
867     TRACE_smpi_comm_in(my_proc_id, "action_scatter", new simgrid::instr::CollTIData(name, args.root, -1.0, args.send_size, args.recv_size,
868                                                                           Datatype::encode(args.datatype1),
869                                                                           Datatype::encode(args.datatype2)));
870
871     Colls::scatter(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
872                   (rank == args.root) ? recv_buffer(args.recv_size * args.datatype2->size()) : nullptr, args.recv_size, args.datatype2, args.root, MPI_COMM_WORLD);
873
874     TRACE_smpi_comm_out(my_proc_id);
875   }
876 };
877
878
879 class ScatterVAction : public ReplayAction<ScatterVArgParser> {
880 public:
881   ScatterVAction() : ReplayAction("scatterV") {}
882   void kernel(simgrid::xbt::ReplayAction& action) override
883   {
884     int rank = MPI_COMM_WORLD->rank();
885     TRACE_smpi_comm_in(my_proc_id, "action_scatterv", new simgrid::instr::VarCollTIData(name, args.root, -1, args.sendcounts, args.recv_size,
886           nullptr, Datatype::encode(args.datatype1),
887           Datatype::encode(args.datatype2)));
888
889     Colls::scatterv((rank == args.root) ? send_buffer(args.send_size_sum * args.datatype1->size()) : nullptr,
890                     args.sendcounts->data(), args.disps.data(), args.datatype1,
891                     recv_buffer(args.recv_size * args.datatype2->size()), args.recv_size, args.datatype2, args.root,
892                     MPI_COMM_WORLD);
893
894     TRACE_smpi_comm_out(my_proc_id);
895   }
896 };
897
898 class ReduceScatterAction : public ReplayAction<ReduceScatterArgParser> {
899 public:
900   ReduceScatterAction() : ReplayAction("reduceScatter") {}
901   void kernel(simgrid::xbt::ReplayAction& action) override
902   {
903     TRACE_smpi_comm_in(my_proc_id, "action_reducescatter",
904                        new simgrid::instr::VarCollTIData("reduceScatter", -1, 0, nullptr, -1, args.recvcounts,
905                                                          std::to_string(args.comp_size), /* ugly hack to print comp_size */
906                                                          Datatype::encode(args.datatype1)));
907
908     Colls::reduce_scatter(send_buffer(args.recv_size_sum * args.datatype1->size()),
909                           recv_buffer(args.recv_size_sum * args.datatype1->size()), args.recvcounts->data(),
910                           args.datatype1, MPI_OP_NULL, MPI_COMM_WORLD);
911
912     smpi_execute_flops(args.comp_size);
913     TRACE_smpi_comm_out(my_proc_id);
914   }
915 };
916
917 class AllToAllVAction : public ReplayAction<AllToAllVArgParser> {
918 public:
919   AllToAllVAction() : ReplayAction("allToAllV") {}
920   void kernel(simgrid::xbt::ReplayAction& action) override
921   {
922     TRACE_smpi_comm_in(my_proc_id, __func__,
923                        new simgrid::instr::VarCollTIData(
924                            "allToAllV", -1, args.send_size_sum, args.sendcounts, args.recv_size_sum, args.recvcounts,
925                            Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
926
927     Colls::alltoallv(send_buffer(args.send_buf_size * args.datatype1->size()), args.sendcounts->data(), args.senddisps.data(), args.datatype1,
928                      recv_buffer(args.recv_buf_size * args.datatype2->size()), args.recvcounts->data(), args.recvdisps.data(), args.datatype2, MPI_COMM_WORLD);
929
930     TRACE_smpi_comm_out(my_proc_id);
931   }
932 };
933 } // Replay Namespace
934 }} // namespace simgrid::smpi
935
936 std::vector<simgrid::smpi::replay::RequestStorage> storage;
937 /** @brief Only initialize the replay, don't do it for real */
938 void smpi_replay_init(int* argc, char*** argv)
939 {
940   simgrid::smpi::Process::init(argc, argv);
941   smpi_process()->mark_as_initialized();
942   smpi_process()->set_replaying(true);
943
944   int my_proc_id = simgrid::s4u::this_actor::get_pid();
945   TRACE_smpi_init(my_proc_id);
946   TRACE_smpi_computing_init(my_proc_id);
947   TRACE_smpi_comm_in(my_proc_id, "smpi_replay_run_init", new simgrid::instr::NoOpTIData("init"));
948   TRACE_smpi_comm_out(my_proc_id);
949   xbt_replay_action_register("init", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::InitAction().execute(action); });
950   xbt_replay_action_register("finalize", [](simgrid::xbt::ReplayAction& action) { /* nothing to do */ });
951   xbt_replay_action_register("comm_size", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
952   xbt_replay_action_register("comm_split",[](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
953   xbt_replay_action_register("comm_dup",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
954
955   xbt_replay_action_register("send",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::SendAction("send").execute(action); });
956   xbt_replay_action_register("Isend", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::SendAction("Isend").execute(action); });
957   xbt_replay_action_register("recv",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::RecvAction("recv").execute(action); });
958   xbt_replay_action_register("Irecv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::RecvAction("Irecv").execute(action); });
959   xbt_replay_action_register("test",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::TestAction().execute(action); });
960   xbt_replay_action_register("wait",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::WaitAction().execute(action); });
961   xbt_replay_action_register("waitAll", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::WaitAllAction().execute(action); });
962   xbt_replay_action_register("barrier", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::BarrierAction().execute(action); });
963   xbt_replay_action_register("bcast",   [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::BcastAction().execute(action); });
964   xbt_replay_action_register("reduce",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ReduceAction().execute(action); });
965   xbt_replay_action_register("allReduce", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllReduceAction().execute(action); });
966   xbt_replay_action_register("allToAll", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllToAllAction().execute(action); });
967   xbt_replay_action_register("allToAllV", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllToAllVAction().execute(action); });
968   xbt_replay_action_register("gather",   [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherAction("gather").execute(action); });
969   xbt_replay_action_register("scatter",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ScatterAction().execute(action); });
970   xbt_replay_action_register("gatherV",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherVAction("gatherV").execute(action); });
971   xbt_replay_action_register("scatterV", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ScatterVAction().execute(action); });
972   xbt_replay_action_register("allGather", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherAction("allGather").execute(action); });
973   xbt_replay_action_register("allGatherV", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherVAction("allGatherV").execute(action); });
974   xbt_replay_action_register("reduceScatter", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ReduceScatterAction().execute(action); });
975   xbt_replay_action_register("compute", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ComputeAction().execute(action); });
976
977   //if we have a delayed start, sleep here.
978   if(*argc>2){
979     double value = xbt_str_parse_double((*argv)[2], "%s is not a double");
980     XBT_VERB("Delayed start for instance - Sleeping for %f flops ",value );
981     smpi_execute_flops(value);
982   } else {
983     //UGLY: force a context switch to be sure that all MSG_processes begin initialization
984     XBT_DEBUG("Force context switch by smpi_execute_flops  - Sleeping for 0.0 flops ");
985     smpi_execute_flops(0.0);
986   }
987 }
988
989 /** @brief actually run the replay after initialization */
990 void smpi_replay_main(int* argc, char*** argv)
991 {
992   static int active_processes = 0;
993   active_processes++;
994   simgrid::xbt::replay_runner(*argc, *argv);
995
996   /* and now, finalize everything */
997   /* One active process will stop. Decrease the counter*/
998   XBT_DEBUG("There are %zu elements in reqq[*]", get_reqq_self()->size());
999   if (not get_reqq_self()->empty()) {
1000     unsigned int count_requests=get_reqq_self()->size();
1001     MPI_Request requests[count_requests];
1002     MPI_Status status[count_requests];
1003     unsigned int i=0;
1004
1005     for (auto const& req : *get_reqq_self()) {
1006       requests[i] = req;
1007       i++;
1008     }
1009     simgrid::smpi::Request::waitall(count_requests, requests, status);
1010   }
1011   delete get_reqq_self();
1012   active_processes--;
1013
1014   if(active_processes==0){
1015     /* Last process alive speaking: end the simulated timer */
1016     XBT_INFO("Simulation time %f", smpi_process()->simulated_elapsed());
1017     smpi_free_replay_tmp_buffers();
1018   }
1019
1020   TRACE_smpi_comm_in(simgrid::s4u::this_actor::get_pid(), "smpi_replay_run_finalize",
1021                      new simgrid::instr::NoOpTIData("finalize"));
1022
1023   smpi_process()->finalize();
1024
1025   TRACE_smpi_comm_out(simgrid::s4u::this_actor::get_pid());
1026   TRACE_smpi_finalize(simgrid::s4u::this_actor::get_pid());
1027 }
1028
1029 /** @brief chain a replay initialization and a replay start */
1030 void smpi_replay_run(int* argc, char*** argv)
1031 {
1032   smpi_replay_init(argc, argv);
1033   smpi_replay_main(argc, argv);
1034 }