src/smpi/include/smpi_replay.hpp

   1 /* Copyright (c) 2009-2023. 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 #ifndef SMPI_REPLAY_HPP_
   6 #define SMPI_REPLAY_HPP_
   7
   8 #include "smpi_actor.hpp"
   9 #include "xbt/ex.h"
  10 #include "xbt/replay.hpp"
  11
  12 #include <boost/algorithm/string/join.hpp>
  13 #include <memory>
  14 #include <sstream>
  15
  16 #define CHECK_ACTION_PARAMS(action, mandatory, optional)                                                               \
  17   {                                                                                                                    \
  18     if ((action).size() < static_cast<unsigned long>((mandatory) + 2)) {                                               \
  19       std::stringstream ss;                                                                                            \
  20       ss << __func__ << " replay failed.\n"                                                                            \
  21          << (action).size() << " items were given on the line. First two should be process_id and action.  "           \
  22          << "This action needs after them " << (mandatory) << " mandatory arguments, and accepts " << (optional)       \
  23          << " optional ones. \n"                                                                                       \
  24          << "The full line that was given is:\n   ";                                                                   \
  25       for (const auto& elem : (action)) {                                                                              \
  26         ss << elem << " ";                                                                                             \
  27       }                                                                                                                \
  28       ss << "\nPlease contact the Simgrid team if support is needed";                                                  \
  29       throw std::invalid_argument(ss.str());                                                                           \
  30     }                                                                                                                  \
  31   }
  32
  33 XBT_PRIVATE unsigned char* smpi_get_tmp_sendbuffer(size_t size);
  34 XBT_PRIVATE unsigned char* smpi_get_tmp_recvbuffer(size_t size);
  35
  36 XBT_PRIVATE void log_timed_action(const simgrid::xbt::ReplayAction& action, double clock);
  37
  38 namespace simgrid::smpi::replay {
  39 extern MPI_Datatype MPI_DEFAULT_TYPE;
  40
  41 class RequestStorage; // Forward decl
  42
  43 /**
  44  * Base class for all parsers.
  45  */
  46 class ActionArgParser {
  47 public:
  48   virtual ~ActionArgParser() = default;
  49   virtual void parse(xbt::ReplayAction& action, const std::string& name) { CHECK_ACTION_PARAMS(action, 0, 0) }
  50 };
  51
  52 class WaitTestParser : public ActionArgParser {
  53 public:
  54   int src;
  55   int dst;
  56   int tag;
  57
  58   void parse(xbt::ReplayAction& action, const std::string& name) override;
  59 };
  60
  61 class SendOrRecvParser : public ActionArgParser {
  62 public:
  63   /* communication partner; if we send, this is the receiver and vice versa */
  64   int partner;
  65   ssize_t size;
  66   int tag;
  67   MPI_Datatype datatype1;
  68
  69   void parse(xbt::ReplayAction& action, const std::string& name) override;
  70 };
  71
  72 class ComputeParser : public ActionArgParser {
  73 public:
  74   double flops;
  75
  76   void parse(xbt::ReplayAction& action, const std::string& name) override;
  77 };
  78
  79 class SleepParser : public ActionArgParser {
  80 public:
  81   double time;
  82
  83   void parse(xbt::ReplayAction& action, const std::string& name) override;
  84 };
  85
  86 class LocationParser : public ActionArgParser {
  87 public:
  88   std::string filename;
  89   int line;
  90
  91   void parse(xbt::ReplayAction& action, const std::string& name) override;
  92 };
  93
  94 class CollCommParser : public ActionArgParser {
  95 public:
  96   size_t size;
  97   double comp_size;
  98   int send_size;
  99   int recv_size;
 100   unsigned comm_size; // size of communicator
 101   int root;
 102   MPI_Datatype datatype1;
 103   MPI_Datatype datatype2;
 104 };
 105
 106 class SendRecvParser : public ActionArgParser {
 107 public:
 108   int dst;
 109   int src;
 110   int sendcount;
 111   int recvcount;
 112   MPI_Datatype datatype1;
 113   MPI_Datatype datatype2;
 114
 115   void parse(xbt::ReplayAction& action, const std::string& name) override;
 116 };
 117
 118 class BcastArgParser : public CollCommParser {
 119 public:
 120   void parse(xbt::ReplayAction& action, const std::string& name) override;
 121 };
 122
 123 class ReduceArgParser : public CollCommParser {
 124 public:
 125   void parse(xbt::ReplayAction& action, const std::string& name) override;
 126 };
 127
 128 class AllReduceArgParser : public CollCommParser {
 129 public:
 130   void parse(xbt::ReplayAction& action, const std::string& name) override;
 131 };
 132
 133 class AllToAllArgParser : public CollCommParser {
 134 public:
 135   void parse(xbt::ReplayAction& action, const std::string& name) override;
 136 };
 137
 138 class GatherArgParser : public CollCommParser {
 139 public:
 140   void parse(xbt::ReplayAction& action, const std::string& name) override;
 141 };
 142
 143 class GatherVArgParser : public CollCommParser {
 144 public:
 145   int recv_size_sum;
 146   std::shared_ptr<std::vector<int>> recvcounts;
 147   std::vector<int> disps;
 148   void parse(xbt::ReplayAction& action, const std::string& name) override;
 149 };
 150
 151 class ScatterArgParser : public CollCommParser {
 152 public:
 153   void parse(xbt::ReplayAction& action, const std::string& name) override;
 154 };
 155
 156 class ScatterVArgParser : public CollCommParser {
 157 public:
 158   int recv_size_sum;
 159   int send_size_sum;
 160   std::shared_ptr<std::vector<int>> sendcounts;
 161   std::vector<int> disps;
 162   void parse(xbt::ReplayAction& action, const std::string& name) override;
 163 };
 164
 165 class ReduceScatterArgParser : public CollCommParser {
 166 public:
 167   int recv_size_sum;
 168   std::shared_ptr<std::vector<int>> recvcounts;
 169   std::vector<int> disps;
 170   void parse(xbt::ReplayAction& action, const std::string& name) override;
 171 };
 172
 173 class ScanArgParser : public CollCommParser {
 174 public:
 175   void parse(xbt::ReplayAction& action, const std::string& name) override;
 176 };
 177
 178 class AllToAllVArgParser : public CollCommParser {
 179 public:
 180   int recv_size_sum;
 181   int send_size_sum;
 182   std::shared_ptr<std::vector<int>> recvcounts;
 183   std::shared_ptr<std::vector<int>> sendcounts;
 184   std::vector<int> senddisps;
 185   std::vector<int> recvdisps;
 186   int send_buf_size;
 187   int recv_buf_size;
 188   void parse(xbt::ReplayAction& action, const std::string& name) override;
 189 };
 190
 191 /**
 192  * Base class for all ReplayActions.
 193  * Note that this class actually implements the behavior of each action
 194  * while the parsing of the replay arguments is done in the @ref ActionArgParser class.
 195  * In other words: The logic goes here, the setup is done by the ActionArgParser.
 196  */
 197 template <class T> class ReplayAction {
 198   const std::string name_;
 199   const aid_t my_proc_id_ = s4u::this_actor::get_pid();
 200   T args_{};
 201
 202 protected:
 203   const std::string& get_name() const { return name_; }
 204   aid_t get_pid() const { return my_proc_id_; }
 205   const T& get_args() const { return args_; }
 206
 207 public:
 208   explicit ReplayAction(const std::string& name) : name_(name) {}
 209   virtual ~ReplayAction() = default;
 210
 211   void execute(xbt::ReplayAction& action)
 212   {
 213     // Needs to be re-initialized for every action, hence here
 214     double start_time = smpi_process()->simulated_elapsed();
 215     args_.parse(action, name_);
 216     kernel(action);
 217     if (name_ != "Init")
 218       log_timed_action(action, start_time);
 219   }
 220
 221   virtual void kernel(simgrid::xbt::ReplayAction& action) = 0;
 222   unsigned char* send_buffer(size_t size) { return smpi_get_tmp_sendbuffer(size); }
 223   unsigned char* recv_buffer(size_t size) { return smpi_get_tmp_recvbuffer(size); }
 224 };
 225
 226 class WaitAction : public ReplayAction<WaitTestParser> {
 227   RequestStorage& req_storage;
 228
 229 public:
 230   explicit WaitAction(RequestStorage& storage) : ReplayAction("Wait"), req_storage(storage) {}
 231   void kernel(xbt::ReplayAction& action) override;
 232 };
 233
 234 class SendAction : public ReplayAction<SendOrRecvParser> {
 235   RequestStorage& req_storage;
 236
 237 public:
 238   explicit SendAction(const std::string& name, RequestStorage& storage) : ReplayAction(name), req_storage(storage) {}
 239   void kernel(xbt::ReplayAction& action) override;
 240 };
 241
 242 class RecvAction : public ReplayAction<SendOrRecvParser> {
 243   RequestStorage& req_storage;
 244
 245 public:
 246   explicit RecvAction(const std::string& name, RequestStorage& storage) : ReplayAction(name), req_storage(storage) {}
 247   void kernel(xbt::ReplayAction& action) override;
 248 };
 249
 250 class ComputeAction : public ReplayAction<ComputeParser> {
 251 public:
 252   explicit ComputeAction() : ReplayAction("compute") {}
 253   void kernel(xbt::ReplayAction& action) override;
 254 };
 255
 256 class SleepAction : public ReplayAction<SleepParser> {
 257 public:
 258   explicit SleepAction() : ReplayAction("sleep") {}
 259   void kernel(xbt::ReplayAction& action) override;
 260 };
 261
 262 class LocationAction : public ReplayAction<LocationParser> {
 263 public:
 264   explicit LocationAction() : ReplayAction("location") {}
 265   void kernel(xbt::ReplayAction& action) override;
 266 };
 267
 268 class TestAction : public ReplayAction<WaitTestParser> {
 269 private:
 270   RequestStorage& req_storage;
 271
 272 public:
 273   explicit TestAction(RequestStorage& storage) : ReplayAction("Test"), req_storage(storage) {}
 274   void kernel(xbt::ReplayAction& action) override;
 275 };
 276
 277 class InitAction : public ReplayAction<ActionArgParser> {
 278 public:
 279   explicit InitAction() : ReplayAction("Init") {}
 280   void kernel(xbt::ReplayAction& action) override;
 281 };
 282
 283 class CommunicatorAction : public ReplayAction<ActionArgParser> {
 284 public:
 285   explicit CommunicatorAction() : ReplayAction("Comm") {}
 286   void kernel(xbt::ReplayAction& action) override;
 287 };
 288
 289 class WaitAllAction : public ReplayAction<ActionArgParser> {
 290   RequestStorage& req_storage;
 291
 292 public:
 293   explicit WaitAllAction(RequestStorage& storage) : ReplayAction("waitall"), req_storage(storage) {}
 294   void kernel(xbt::ReplayAction& action) override;
 295 };
 296
 297 class SendRecvAction : public ReplayAction<SendRecvParser> {
 298 public:
 299   explicit SendRecvAction() : ReplayAction("sendrecv") {}
 300   void kernel(xbt::ReplayAction& action) override;
 301 };
 302
 303 class BarrierAction : public ReplayAction<ActionArgParser> {
 304 public:
 305   explicit BarrierAction() : ReplayAction("barrier") {}
 306   void kernel(xbt::ReplayAction& action) override;
 307 };
 308
 309 class BcastAction : public ReplayAction<BcastArgParser> {
 310 public:
 311   explicit BcastAction() : ReplayAction("bcast") {}
 312   void kernel(xbt::ReplayAction& action) override;
 313 };
 314
 315 class ReduceAction : public ReplayAction<ReduceArgParser> {
 316 public:
 317   explicit ReduceAction() : ReplayAction("reduce") {}
 318   void kernel(xbt::ReplayAction& action) override;
 319 };
 320
 321 class AllReduceAction : public ReplayAction<AllReduceArgParser> {
 322 public:
 323   explicit AllReduceAction() : ReplayAction("allreduce") {}
 324   void kernel(xbt::ReplayAction& action) override;
 325 };
 326
 327 class AllToAllAction : public ReplayAction<AllToAllArgParser> {
 328 public:
 329   explicit AllToAllAction() : ReplayAction("alltoall") {}
 330   void kernel(xbt::ReplayAction& action) override;
 331 };
 332
 333 class GatherAction : public ReplayAction<GatherArgParser> {
 334 public:
 335   using ReplayAction::ReplayAction;
 336   void kernel(xbt::ReplayAction& action) override;
 337 };
 338
 339 class GatherVAction : public ReplayAction<GatherVArgParser> {
 340 public:
 341   using ReplayAction::ReplayAction;
 342   void kernel(xbt::ReplayAction& action) override;
 343 };
 344
 345 class ScatterAction : public ReplayAction<ScatterArgParser> {
 346 public:
 347   explicit ScatterAction() : ReplayAction("scatter") {}
 348   void kernel(xbt::ReplayAction& action) override;
 349 };
 350
 351 class ScatterVAction : public ReplayAction<ScatterVArgParser> {
 352 public:
 353   explicit ScatterVAction() : ReplayAction("scatterv") {}
 354   void kernel(xbt::ReplayAction& action) override;
 355 };
 356
 357 class ReduceScatterAction : public ReplayAction<ReduceScatterArgParser> {
 358 public:
 359   explicit ReduceScatterAction() : ReplayAction("reducescatter") {}
 360   void kernel(xbt::ReplayAction& action) override;
 361 };
 362
 363 class ScanAction : public ReplayAction<ScanArgParser> {
 364 public:
 365   using ReplayAction::ReplayAction;
 366   void kernel(xbt::ReplayAction& action) override;
 367 };
 368
 369 class AllToAllVAction : public ReplayAction<AllToAllVArgParser> {
 370 public:
 371   explicit AllToAllVAction() : ReplayAction("alltoallv") {}
 372   void kernel(xbt::ReplayAction& action) override;
 373 };
 374
 375 } // namespace simgrid::smpi::replay
 376
 377 #endif