Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
[TRACING] Rename TIT action allGather -> allgather
[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 "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(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(src, dst, tag), MPI_REQUEST_NULL});
138     }
139 };
140
141 void WaitTestParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
142 {
143   CHECK_ACTION_PARAMS(action, 3, 0)
144   src = std::stoi(action[2]);
145   dst = std::stoi(action[3]);
146   tag = std::stoi(action[4]);
147 }
148
149 void SendRecvParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
150 {
151   CHECK_ACTION_PARAMS(action, 3, 1)
152   partner = std::stoi(action[2]);
153   tag     = std::stoi(action[3]);
154   size    = parse_double(action[4]);
155   if (action.size() > 5)
156     datatype1 = simgrid::smpi::Datatype::decode(action[5]);
157 }
158
159 void ComputeParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
160 {
161   CHECK_ACTION_PARAMS(action, 1, 0)
162   flops = parse_double(action[2]);
163 }
164
165 void BcastArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
166 {
167   CHECK_ACTION_PARAMS(action, 1, 2)
168   size = parse_double(action[2]);
169   root = (action.size() > 3) ? std::stoi(action[3]) : 0;
170   if (action.size() > 4)
171     datatype1 = simgrid::smpi::Datatype::decode(action[4]);
172 }
173
174 void ReduceArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
175 {
176   CHECK_ACTION_PARAMS(action, 2, 2)
177   comm_size = parse_double(action[2]);
178   comp_size = parse_double(action[3]);
179   root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
180   if (action.size() > 5)
181     datatype1 = simgrid::smpi::Datatype::decode(action[5]);
182 }
183
184 void AllReduceArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
185 {
186   CHECK_ACTION_PARAMS(action, 2, 1)
187   comm_size = parse_double(action[2]);
188   comp_size = parse_double(action[3]);
189   if (action.size() > 4)
190     datatype1 = simgrid::smpi::Datatype::decode(action[4]);
191 }
192
193 void AllToAllArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
194 {
195   CHECK_ACTION_PARAMS(action, 2, 1)
196   comm_size = MPI_COMM_WORLD->size();
197   send_size = parse_double(action[2]);
198   recv_size = parse_double(action[3]);
199
200   if (action.size() > 4)
201     datatype1 = simgrid::smpi::Datatype::decode(action[4]);
202   if (action.size() > 5)
203     datatype2 = simgrid::smpi::Datatype::decode(action[5]);
204 }
205
206 void GatherArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
207 {
208   /* The structure of the gather action for the rank 0 (total 4 processes) is the following:
209         0 gather 68 68 0 0 0
210       where:
211         1) 68 is the sendcounts
212         2) 68 is the recvcounts
213         3) 0 is the root node
214         4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
215         5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
216   */
217   CHECK_ACTION_PARAMS(action, 2, 3)
218   comm_size = MPI_COMM_WORLD->size();
219   send_size = parse_double(action[2]);
220   recv_size = parse_double(action[3]);
221
222   if (name == "gather") {
223     root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
224     if (action.size() > 5)
225       datatype1 = simgrid::smpi::Datatype::decode(action[5]);
226     if (action.size() > 6)
227       datatype2 = simgrid::smpi::Datatype::decode(action[6]);
228   } else {
229     if (action.size() > 4)
230       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
231     if (action.size() > 5)
232       datatype2 = simgrid::smpi::Datatype::decode(action[5]);
233   }
234 }
235
236 void GatherVArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
237 {
238   /* The structure of the gatherv action for the rank 0 (total 4 processes) is the following:
239        0 gather 68 68 10 10 10 0 0 0
240      where:
241        1) 68 is the sendcount
242        2) 68 10 10 10 is the recvcounts
243        3) 0 is the root node
244        4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
245        5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
246   */
247   comm_size = MPI_COMM_WORLD->size();
248   CHECK_ACTION_PARAMS(action, comm_size + 1, 2)
249   send_size  = parse_double(action[2]);
250   disps      = std::vector<int>(comm_size, 0);
251   recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
252
253   if (name == "gatherv") {
254     root = (action.size() > 3 + comm_size) ? std::stoi(action[3 + comm_size]) : 0;
255     if (action.size() > 4 + comm_size)
256       datatype1 = simgrid::smpi::Datatype::decode(action[4 + comm_size]);
257     if (action.size() > 5 + comm_size)
258       datatype2 = simgrid::smpi::Datatype::decode(action[5 + comm_size]);
259   } else {
260     int datatype_index = 0;
261     int disp_index     = 0;
262     /* The 3 comes from "0 gather <sendcount>", which must always be present.
263      * The + comm_size is the recvcounts array, which must also be present
264      */
265     if (action.size() > 3 + comm_size + comm_size) { /* datatype + disp are specified */
266       datatype_index = 3 + comm_size;
267       disp_index     = datatype_index + 1;
268       datatype1      = simgrid::smpi::Datatype::decode(action[datatype_index]);
269       datatype2      = simgrid::smpi::Datatype::decode(action[datatype_index]);
270     } else if (action.size() >
271                3 + comm_size + 2) { /* disps specified; datatype is not specified; use the default one */
272       disp_index = 3 + comm_size;
273     } else if (action.size() > 3 + comm_size) { /* only datatype, no disp specified */
274       datatype_index = 3 + comm_size;
275       datatype1      = simgrid::smpi::Datatype::decode(action[datatype_index]);
276       datatype2      = simgrid::smpi::Datatype::decode(action[datatype_index]);
277     }
278
279     if (disp_index != 0) {
280       for (unsigned int i = 0; i < comm_size; i++)
281         disps[i]          = std::stoi(action[disp_index + i]);
282     }
283   }
284
285   for (unsigned int i = 0; i < comm_size; i++) {
286     (*recvcounts)[i] = std::stoi(action[i + 3]);
287   }
288   recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
289 }
290
291 void ScatterArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
292 {
293   /* The structure of the scatter action for the rank 0 (total 4 processes) is the following:
294         0 gather 68 68 0 0 0
295       where:
296         1) 68 is the sendcounts
297         2) 68 is the recvcounts
298         3) 0 is the root node
299         4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
300         5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
301   */
302   CHECK_ACTION_PARAMS(action, 2, 3)
303   comm_size = MPI_COMM_WORLD->size();
304   send_size = parse_double(action[2]);
305   recv_size = parse_double(action[3]);
306   root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
307   if (action.size() > 5)
308     datatype1 = simgrid::smpi::Datatype::decode(action[5]);
309   if (action.size() > 6)
310     datatype2 = simgrid::smpi::Datatype::decode(action[6]);
311 }
312
313 void ScatterVArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
314 {
315   /* The structure of the scatterv action for the rank 0 (total 4 processes) is the following:
316      0 gather 68 10 10 10 68 0 0 0
317       where:
318       1) 68 10 10 10 is the sendcounts
319       2) 68 is the recvcount
320       3) 0 is the root node
321       4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
322       5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
323   */
324   CHECK_ACTION_PARAMS(action, comm_size + 1, 2)
325   recv_size  = parse_double(action[2 + comm_size]);
326   disps      = std::vector<int>(comm_size, 0);
327   sendcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
328
329   if (action.size() > 5 + comm_size)
330     datatype1 = simgrid::smpi::Datatype::decode(action[4 + comm_size]);
331   if (action.size() > 5 + comm_size)
332     datatype2 = simgrid::smpi::Datatype::decode(action[5]);
333
334   for (unsigned int i = 0; i < comm_size; i++) {
335     (*sendcounts)[i] = std::stoi(action[i + 2]);
336   }
337   send_size_sum = std::accumulate(sendcounts->begin(), sendcounts->end(), 0);
338   root          = (action.size() > 3 + comm_size) ? std::stoi(action[3 + comm_size]) : 0;
339 }
340
341 void ReduceScatterArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
342 {
343   /* The structure of the reducescatter action for the rank 0 (total 4 processes) is the following:
344        0 reducescatter 275427 275427 275427 204020 11346849 0
345      where:
346        1) The first four values after the name of the action declare the recvcounts array
347        2) The value 11346849 is the amount of instructions
348        3) The last value corresponds to the datatype, see simgrid::smpi::Datatype::decode().
349   */
350   comm_size = MPI_COMM_WORLD->size();
351   CHECK_ACTION_PARAMS(action, comm_size + 1, 1)
352   comp_size  = parse_double(action[2 + comm_size]);
353   recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
354   if (action.size() > 3 + comm_size)
355     datatype1 = simgrid::smpi::Datatype::decode(action[3 + comm_size]);
356
357   for (unsigned int i = 0; i < comm_size; i++) {
358     recvcounts->push_back(std::stoi(action[i + 2]));
359   }
360   recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
361 }
362
363 void AllToAllVArgParser::parse(simgrid::xbt::ReplayAction& action, std::string name)
364 {
365   /* The structure of the allToAllV action for the rank 0 (total 4 processes) is the following:
366         0 allToAllV 100 1 7 10 12 100 1 70 10 5
367      where:
368       1) 100 is the size of the send buffer *sizeof(int),
369       2) 1 7 10 12 is the sendcounts array
370       3) 100*sizeof(int) is the size of the receiver buffer
371       4)  1 70 10 5 is the recvcounts array
372   */
373   comm_size = MPI_COMM_WORLD->size();
374   CHECK_ACTION_PARAMS(action, 2 * comm_size + 2, 2)
375   sendcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
376   recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
377   senddisps  = std::vector<int>(comm_size, 0);
378   recvdisps  = std::vector<int>(comm_size, 0);
379
380   if (action.size() > 5 + 2 * comm_size)
381     datatype1 = simgrid::smpi::Datatype::decode(action[4 + 2 * comm_size]);
382   if (action.size() > 5 + 2 * comm_size)
383     datatype2 = simgrid::smpi::Datatype::decode(action[5 + 2 * comm_size]);
384
385   send_buf_size = parse_double(action[2]);
386   recv_buf_size = parse_double(action[3 + comm_size]);
387   for (unsigned int i = 0; i < comm_size; i++) {
388     (*sendcounts)[i] = std::stoi(action[3 + i]);
389     (*recvcounts)[i] = std::stoi(action[4 + comm_size + i]);
390   }
391   send_size_sum = std::accumulate(sendcounts->begin(), sendcounts->end(), 0);
392   recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
393 }
394
395 void WaitAction::kernel(simgrid::xbt::ReplayAction& action)
396 {
397   std::string s = boost::algorithm::join(action, " ");
398   xbt_assert(req_storage.size(), "action wait not preceded by any irecv or isend: %s", s.c_str());
399   MPI_Request request = req_storage.find(args.src, args.dst, args.tag);
400   req_storage.remove(request);
401
402   if (request == MPI_REQUEST_NULL) {
403     /* Assume that the trace is well formed, meaning the comm might have been caught by a MPI_test. Then just
404      * return.*/
405     return;
406   }
407
408   int rank = request->comm() != MPI_COMM_NULL ? request->comm()->rank() : -1;
409
410   // Must be taken before Request::wait() since the request may be set to
411   // MPI_REQUEST_NULL by Request::wait!
412   bool is_wait_for_receive = (request->flags() & MPI_REQ_RECV);
413   // TODO: Here we take the rank while we normally take the process id (look for my_proc_id)
414   TRACE_smpi_comm_in(rank, __func__, new simgrid::instr::WaitTIData(args.src, args.dst, args.tag));
415
416   MPI_Status status;
417   Request::wait(&request, &status);
418
419   TRACE_smpi_comm_out(rank);
420   if (is_wait_for_receive)
421     TRACE_smpi_recv(args.src, args.dst, args.tag);
422 }
423
424 void SendAction::kernel(simgrid::xbt::ReplayAction& action)
425 {
426   int dst_traced = MPI_COMM_WORLD->group()->actor(args.partner)->get_pid();
427
428   TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData(name, args.partner, args.size,
429         args.tag, Datatype::encode(args.datatype1)));
430   if (not TRACE_smpi_view_internals())
431     TRACE_smpi_send(my_proc_id, my_proc_id, dst_traced, args.tag, args.size * args.datatype1->size());
432
433   if (name == "send") {
434     Request::send(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
435   } else if (name == "Isend") {
436     MPI_Request request = Request::isend(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
437     req_storage.add(request);
438   } else {
439     xbt_die("Don't know this action, %s", name.c_str());
440   }
441
442   TRACE_smpi_comm_out(my_proc_id);
443 }
444
445 void RecvAction::kernel(simgrid::xbt::ReplayAction& action)
446 {
447   int src_traced = MPI_COMM_WORLD->group()->actor(args.partner)->get_pid();
448
449   TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData(name, args.partner, args.size,
450         args.tag, Datatype::encode(args.datatype1)));
451
452   MPI_Status status;
453   // unknown size from the receiver point of view
454   if (args.size <= 0.0) {
455     Request::probe(args.partner, args.tag, MPI_COMM_WORLD, &status);
456     args.size = status.count;
457   }
458
459   if (name == "recv") {
460     Request::recv(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD, &status);
461   } else if (name == "Irecv") {
462     MPI_Request request = Request::irecv(nullptr, args.size, args.datatype1, args.partner, args.tag, MPI_COMM_WORLD);
463     req_storage.add(request);
464   }
465
466   TRACE_smpi_comm_out(my_proc_id);
467   // TODO: Check why this was only activated in the "recv" case and not in the "Irecv" case
468   if (name == "recv" && not TRACE_smpi_view_internals()) {
469     TRACE_smpi_recv(src_traced, my_proc_id, args.tag);
470   }
471 }
472
473 void ComputeAction::kernel(simgrid::xbt::ReplayAction& action)
474 {
475   TRACE_smpi_computing_in(my_proc_id, args.flops);
476   smpi_execute_flops(args.flops);
477   TRACE_smpi_computing_out(my_proc_id);
478 }
479
480 void TestAction::kernel(simgrid::xbt::ReplayAction& action)
481 {
482   MPI_Request request = req_storage.find(args.src, args.dst, args.tag);
483   req_storage.remove(request);
484   // if request is null here, this may mean that a previous test has succeeded
485   // Different times in traced application and replayed version may lead to this
486   // In this case, ignore the extra calls.
487   if (request != MPI_REQUEST_NULL) {
488     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::NoOpTIData("test"));
489
490     MPI_Status status;
491     int flag = Request::test(&request, &status);
492
493     XBT_DEBUG("MPI_Test result: %d", flag);
494     /* push back request in vector to be caught by a subsequent wait. if the test did succeed, the request is now
495      * nullptr.*/
496     if (request == MPI_REQUEST_NULL)
497       req_storage.addNullRequest(args.src, args.dst, args.tag);
498     else
499       req_storage.add(request);
500
501     TRACE_smpi_comm_out(my_proc_id);
502   }
503 }
504
505 void InitAction::kernel(simgrid::xbt::ReplayAction& action)
506 {
507   CHECK_ACTION_PARAMS(action, 0, 1)
508     MPI_DEFAULT_TYPE = (action.size() > 2) ? MPI_DOUBLE // default MPE datatype
509     : MPI_BYTE;  // default TAU datatype
510
511   /* start a simulated timer */
512   smpi_process()->simulated_start();
513 }
514
515 void CommunicatorAction::kernel(simgrid::xbt::ReplayAction& action)
516 {
517   /* nothing to do */
518 }
519
520 void WaitAllAction::kernel(simgrid::xbt::ReplayAction& action)
521 {
522   const unsigned int count_requests = req_storage.size();
523
524   if (count_requests > 0) {
525     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData("waitall", -1, count_requests, ""));
526     std::vector<std::pair</*sender*/int,/*recv*/int>> sender_receiver;
527     std::vector<MPI_Request> reqs;
528     req_storage.get_requests(reqs);
529     for (const auto& req : reqs) {
530       if (req && (req->flags() & MPI_REQ_RECV)) {
531         sender_receiver.push_back({req->src(), req->dst()});
532       }
533     }
534     MPI_Status status[count_requests];
535     Request::waitall(count_requests, &(reqs.data())[0], status);
536     req_storage.get_store().clear();
537
538     for (auto& pair : sender_receiver) {
539       TRACE_smpi_recv(pair.first, pair.second, 0);
540     }
541     TRACE_smpi_comm_out(my_proc_id);
542   }
543 }
544
545 void BarrierAction::kernel(simgrid::xbt::ReplayAction& action)
546 {
547   TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::NoOpTIData("barrier"));
548   Colls::barrier(MPI_COMM_WORLD);
549   TRACE_smpi_comm_out(my_proc_id);
550 }
551
552 void BcastAction::kernel(simgrid::xbt::ReplayAction& action)
553 {
554   TRACE_smpi_comm_in(my_proc_id, "action_bcast",
555       new simgrid::instr::CollTIData("bcast", MPI_COMM_WORLD->group()->actor(args.root)->get_pid(),
556         -1.0, args.size, -1, Datatype::encode(args.datatype1), ""));
557
558   Colls::bcast(send_buffer(args.size * args.datatype1->size()), args.size, args.datatype1, args.root, MPI_COMM_WORLD);
559
560   TRACE_smpi_comm_out(my_proc_id);
561 }
562
563 void ReduceAction::kernel(simgrid::xbt::ReplayAction& action)
564 {
565   TRACE_smpi_comm_in(my_proc_id, "action_reduce",
566       new simgrid::instr::CollTIData("reduce", MPI_COMM_WORLD->group()->actor(args.root)->get_pid(),
567         args.comp_size, args.comm_size, -1,
568         Datatype::encode(args.datatype1), ""));
569
570   Colls::reduce(send_buffer(args.comm_size * args.datatype1->size()),
571       recv_buffer(args.comm_size * args.datatype1->size()), args.comm_size, args.datatype1, MPI_OP_NULL, args.root, MPI_COMM_WORLD);
572   smpi_execute_flops(args.comp_size);
573
574   TRACE_smpi_comm_out(my_proc_id);
575 }
576
577 void AllReduceAction::kernel(simgrid::xbt::ReplayAction& action)
578 {
579   TRACE_smpi_comm_in(my_proc_id, "action_allreduce", new simgrid::instr::CollTIData("allreduce", -1, args.comp_size, args.comm_size, -1,
580         Datatype::encode(args.datatype1), ""));
581
582   Colls::allreduce(send_buffer(args.comm_size * args.datatype1->size()),
583       recv_buffer(args.comm_size * args.datatype1->size()), args.comm_size, args.datatype1, MPI_OP_NULL, MPI_COMM_WORLD);
584   smpi_execute_flops(args.comp_size);
585
586   TRACE_smpi_comm_out(my_proc_id);
587 }
588
589 void AllToAllAction::kernel(simgrid::xbt::ReplayAction& action)
590 {
591   TRACE_smpi_comm_in(my_proc_id, "action_allToAll",
592       new simgrid::instr::CollTIData("allToAll", -1, -1.0, args.send_size, args.recv_size,
593         Datatype::encode(args.datatype1),
594         Datatype::encode(args.datatype2)));
595
596   Colls::alltoall(send_buffer(args.send_size * args.comm_size * args.datatype1->size()), args.send_size,
597       args.datatype1, recv_buffer(args.recv_size * args.comm_size * args.datatype2->size()),
598       args.recv_size, args.datatype2, MPI_COMM_WORLD);
599
600   TRACE_smpi_comm_out(my_proc_id);
601 }
602
603 void GatherAction::kernel(simgrid::xbt::ReplayAction& action)
604 {
605   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,
606         Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
607
608   if (name == "gather") {
609     int rank = MPI_COMM_WORLD->rank();
610     Colls::gather(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
611         (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);
612   }
613   else
614     Colls::allgather(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
615         recv_buffer(args.recv_size * args.datatype2->size()), args.recv_size, args.datatype2, MPI_COMM_WORLD);
616
617   TRACE_smpi_comm_out(my_proc_id);
618 }
619
620 void GatherVAction::kernel(simgrid::xbt::ReplayAction& action)
621 {
622   int rank = MPI_COMM_WORLD->rank();
623
624   TRACE_smpi_comm_in(my_proc_id, name.c_str(), new simgrid::instr::VarCollTIData(
625         name, (name == "gatherv") ? args.root : -1, args.send_size, nullptr, -1, args.recvcounts,
626         Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
627
628   if (name == "gatherv") {
629     Colls::gatherv(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
630         (rank == args.root) ? recv_buffer(args.recv_size_sum * args.datatype2->size()) : nullptr,
631         args.recvcounts->data(), args.disps.data(), args.datatype2, args.root, MPI_COMM_WORLD);
632   }
633   else {
634     Colls::allgatherv(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
635         recv_buffer(args.recv_size_sum * args.datatype2->size()), args.recvcounts->data(),
636         args.disps.data(), args.datatype2, MPI_COMM_WORLD);
637   }
638
639   TRACE_smpi_comm_out(my_proc_id);
640 }
641
642 void ScatterAction::kernel(simgrid::xbt::ReplayAction& action)
643 {
644   int rank = MPI_COMM_WORLD->rank();
645   TRACE_smpi_comm_in(my_proc_id, "action_scatter", new simgrid::instr::CollTIData(name, args.root, -1.0, args.send_size, args.recv_size,
646         Datatype::encode(args.datatype1),
647         Datatype::encode(args.datatype2)));
648
649   Colls::scatter(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
650       (rank == args.root) ? recv_buffer(args.recv_size * args.datatype2->size()) : nullptr, args.recv_size, args.datatype2, args.root, MPI_COMM_WORLD);
651
652   TRACE_smpi_comm_out(my_proc_id);
653 }
654
655 void ScatterVAction::kernel(simgrid::xbt::ReplayAction& action)
656 {
657   int rank = MPI_COMM_WORLD->rank();
658   TRACE_smpi_comm_in(my_proc_id, "action_scatterv", new simgrid::instr::VarCollTIData(name, args.root, -1, args.sendcounts, args.recv_size,
659         nullptr, Datatype::encode(args.datatype1),
660         Datatype::encode(args.datatype2)));
661
662   Colls::scatterv((rank == args.root) ? send_buffer(args.send_size_sum * args.datatype1->size()) : nullptr,
663       args.sendcounts->data(), args.disps.data(), args.datatype1,
664       recv_buffer(args.recv_size * args.datatype2->size()), args.recv_size, args.datatype2, args.root,
665       MPI_COMM_WORLD);
666
667   TRACE_smpi_comm_out(my_proc_id);
668 }
669
670 void ReduceScatterAction::kernel(simgrid::xbt::ReplayAction& action)
671 {
672   TRACE_smpi_comm_in(my_proc_id, "action_reducescatter",
673       new simgrid::instr::VarCollTIData("reducescatter", -1, 0, nullptr, -1, args.recvcounts,
674         std::to_string(args.comp_size), /* ugly hack to print comp_size */
675         Datatype::encode(args.datatype1)));
676
677   Colls::reduce_scatter(send_buffer(args.recv_size_sum * args.datatype1->size()),
678       recv_buffer(args.recv_size_sum * args.datatype1->size()), args.recvcounts->data(),
679       args.datatype1, MPI_OP_NULL, MPI_COMM_WORLD);
680
681   smpi_execute_flops(args.comp_size);
682   TRACE_smpi_comm_out(my_proc_id);
683 }
684
685 void AllToAllVAction::kernel(simgrid::xbt::ReplayAction& action)
686 {
687   TRACE_smpi_comm_in(my_proc_id, __func__,
688       new simgrid::instr::VarCollTIData(
689         "allToAllV", -1, args.send_size_sum, args.sendcounts, args.recv_size_sum, args.recvcounts,
690         Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
691
692   Colls::alltoallv(send_buffer(args.send_buf_size * args.datatype1->size()), args.sendcounts->data(), args.senddisps.data(), args.datatype1,
693       recv_buffer(args.recv_buf_size * args.datatype2->size()), args.recvcounts->data(), args.recvdisps.data(), args.datatype2, MPI_COMM_WORLD);
694
695   TRACE_smpi_comm_out(my_proc_id);
696 }
697 } // Replay Namespace
698 }} // namespace simgrid::smpi
699
700 static std::vector<simgrid::smpi::replay::RequestStorage> storage;
701 /** @brief Only initialize the replay, don't do it for real */
702 void smpi_replay_init(int* argc, char*** argv)
703 {
704   simgrid::smpi::Process::init(argc, argv);
705   smpi_process()->mark_as_initialized();
706   smpi_process()->set_replaying(true);
707
708   int my_proc_id = simgrid::s4u::this_actor::get_pid();
709   storage.resize(smpi_process_count());
710
711   TRACE_smpi_init(my_proc_id);
712   TRACE_smpi_computing_init(my_proc_id);
713   TRACE_smpi_comm_in(my_proc_id, "smpi_replay_run_init", new simgrid::instr::NoOpTIData("init"));
714   TRACE_smpi_comm_out(my_proc_id);
715   xbt_replay_action_register("init", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::InitAction().execute(action); });
716   xbt_replay_action_register("finalize", [](simgrid::xbt::ReplayAction& action) { /* nothing to do */ });
717   xbt_replay_action_register("comm_size", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
718   xbt_replay_action_register("comm_split",[](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
719   xbt_replay_action_register("comm_dup",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::CommunicatorAction().execute(action); });
720   xbt_replay_action_register("send",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::SendAction("send", storage[simgrid::s4u::this_actor::get_pid()-1]).execute(action); });
721   xbt_replay_action_register("Isend", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::SendAction("Isend", storage[simgrid::s4u::this_actor::get_pid()-1]).execute(action); });
722   xbt_replay_action_register("recv",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::RecvAction("recv", storage[simgrid::s4u::this_actor::get_pid()-1]).execute(action); });
723   xbt_replay_action_register("Irecv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::RecvAction("Irecv", storage[simgrid::s4u::this_actor::get_pid()-1]).execute(action); });
724   xbt_replay_action_register("test",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::TestAction(storage[simgrid::s4u::this_actor::get_pid()-1]).execute(action); });
725   xbt_replay_action_register("wait",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::WaitAction(storage[simgrid::s4u::this_actor::get_pid()-1]).execute(action); });
726   xbt_replay_action_register("waitall", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::WaitAllAction(storage[simgrid::s4u::this_actor::get_pid()-1]).execute(action); });
727   xbt_replay_action_register("barrier", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::BarrierAction().execute(action); });
728   xbt_replay_action_register("bcast",   [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::BcastAction().execute(action); });
729   xbt_replay_action_register("reduce",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ReduceAction().execute(action); });
730   xbt_replay_action_register("allreduce", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllReduceAction().execute(action); });
731   xbt_replay_action_register("allToAll", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllToAllAction().execute(action); });
732   xbt_replay_action_register("allToAllV", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::AllToAllVAction().execute(action); });
733   xbt_replay_action_register("gather",   [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherAction("gather").execute(action); });
734   xbt_replay_action_register("scatter",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ScatterAction().execute(action); });
735   xbt_replay_action_register("gatherv",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherVAction("gatherv").execute(action); });
736   xbt_replay_action_register("scatterv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ScatterVAction().execute(action); });
737   xbt_replay_action_register("allgather", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherAction("allgather").execute(action); });
738   xbt_replay_action_register("allgatherv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::GatherVAction("allgatherv").execute(action); });
739   xbt_replay_action_register("reducescatter", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ReduceScatterAction().execute(action); });
740   xbt_replay_action_register("compute", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::replay::ComputeAction().execute(action); });
741
742   //if we have a delayed start, sleep here.
743   if(*argc>2){
744     double value = xbt_str_parse_double((*argv)[2], "%s is not a double");
745     XBT_VERB("Delayed start for instance - Sleeping for %f flops ",value );
746     smpi_execute_flops(value);
747   } else {
748     //UGLY: force a context switch to be sure that all MSG_processes begin initialization
749     XBT_DEBUG("Force context switch by smpi_execute_flops  - Sleeping for 0.0 flops ");
750     smpi_execute_flops(0.0);
751   }
752 }
753
754 /** @brief actually run the replay after initialization */
755 void smpi_replay_main(int* argc, char*** argv)
756 {
757   static int active_processes = 0;
758   active_processes++;
759   simgrid::xbt::replay_runner(*argc, *argv);
760
761   /* and now, finalize everything */
762   /* One active process will stop. Decrease the counter*/
763   unsigned int count_requests = storage[simgrid::s4u::this_actor::get_pid() - 1].size();
764   XBT_DEBUG("There are %ud elements in reqq[*]", count_requests);
765   if (count_requests > 0) {
766     MPI_Request requests[count_requests];
767     MPI_Status status[count_requests];
768     unsigned int i=0;
769
770     for (auto const& pair : storage[simgrid::s4u::this_actor::get_pid() - 1].get_store()) {
771       requests[i] = pair.second;
772       i++;
773     }
774     simgrid::smpi::Request::waitall(count_requests, requests, status);
775   }
776   active_processes--;
777
778   if(active_processes==0){
779     /* Last process alive speaking: end the simulated timer */
780     XBT_INFO("Simulation time %f", smpi_process()->simulated_elapsed());
781     smpi_free_replay_tmp_buffers();
782   }
783
784   TRACE_smpi_comm_in(simgrid::s4u::this_actor::get_pid(), "smpi_replay_run_finalize",
785                      new simgrid::instr::NoOpTIData("finalize"));
786
787   smpi_process()->finalize();
788
789   TRACE_smpi_comm_out(simgrid::s4u::this_actor::get_pid());
790   TRACE_smpi_finalize(simgrid::s4u::this_actor::get_pid());
791 }
792
793 /** @brief chain a replay initialization and a replay start */
794 void smpi_replay_run(int* argc, char*** argv)
795 {
796   smpi_replay_init(argc, argv);
797   smpi_replay_main(argc, argv);
798 }