src/smpi/internals/smpi_bench.cpp

   1 /* Copyright (c) 2007-2021. 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 "getopt.h"
   7 #include "private.hpp"
   8 #include "simgrid/host.h"
   9 #include "simgrid/modelchecker.h"
  10 #include "simgrid/s4u/Exec.hpp"
  11 #include "smpi_comm.hpp"
  12 #include "smpi_utils.hpp"
  13 #include "src/internal_config.h"
  14 #include "src/mc/mc_replay.hpp"
  15 #include "xbt/config.hpp"
  16 #include "xbt/file.hpp"
  17
  18 #include "src/smpi/include/smpi_actor.hpp"
  19 #include <unordered_map>
  20
  21 #ifndef WIN32
  22 #include <sys/mman.h>
  23 #endif
  24 #include <cerrno>
  25 #include <cmath>
  26
  27 #if HAVE_PAPI
  28 #include <papi.h>
  29 #endif
  30
  31 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_bench, smpi, "Logging specific to SMPI (benchmarking)");
  32
  33 static simgrid::config::Flag<double>
  34     smpi_wtime_sleep("smpi/wtime",
  35                      "Minimum time to inject inside a call to MPI_Wtime(), gettimeofday() and clock_gettime()",
  36                      1e-8 /* Documented to be 10 ns */);
  37
  38 // Private execute_flops used by smpi_execute and smpi_execute_benched
  39 void private_execute_flops(double flops) {
  40   xbt_assert(flops >= 0, "You're trying to execute a negative amount of flops (%f)!", flops);
  41   XBT_DEBUG("Handle real computation time: %f flops", flops);
  42   simgrid::s4u::this_actor::exec_init(flops)
  43       ->set_name("computation")
  44       ->set_tracing_category(smpi_process()->get_tracing_category())
  45       ->start()
  46       ->wait();
  47   smpi_switch_data_segment(simgrid::s4u::Actor::self());
  48 }
  49
  50 void smpi_execute_flops(double flops)
  51 {
  52   private_execute_flops(flops);
  53 }
  54
  55 void smpi_execute(double duration)
  56 {
  57   if (duration >= smpi_cfg_cpu_thresh()) {
  58     XBT_DEBUG("Sleep for %g to handle real computation time", duration);
  59     private_execute_flops(duration * smpi_cfg_host_speed());
  60   } else {
  61     XBT_DEBUG("Real computation took %g while option smpi/cpu-threshold is set to %g => ignore it", duration,
  62               smpi_cfg_cpu_thresh());
  63   }
  64 }
  65
  66 void smpi_execute_benched(double duration)
  67 {
  68   smpi_bench_end();
  69   double speed = sg_host_get_speed(sg_host_self());
  70   smpi_execute_flops(duration*speed);
  71   smpi_bench_begin();
  72 }
  73
  74 void smpi_execute_flops_benched(double flops) {
  75   smpi_bench_end();
  76   smpi_execute_flops(flops);
  77   smpi_bench_begin();
  78 }
  79
  80 void smpi_bench_begin()
  81 {
  82   if (smpi_cfg_privatization() == SmpiPrivStrategies::MMAP) {
  83     smpi_switch_data_segment(simgrid::s4u::Actor::self());
  84   }
  85
  86   if (MC_is_active() || MC_record_replay_is_active())
  87     return;
  88
  89 #if HAVE_PAPI
  90   if (not smpi_cfg_papi_events_file().empty()) {
  91     int event_set = smpi_process()->papi_event_set();
  92     // PAPI_start sets everything to 0! See man(3) PAPI_start
  93     if (PAPI_LOW_LEVEL_INITED == PAPI_is_initialized() && event_set)
  94       xbt_assert(PAPI_start(event_set) == PAPI_OK,
  95                  "Could not start PAPI counters (TODO: this needs some proper handling).");
  96   }
  97 #endif
  98   xbt_os_threadtimer_start(smpi_process()->timer());
  99 }
 100
 101 double smpi_adjust_comp_speed(){
 102   double speedup=1;
 103   if (smpi_cfg_comp_adjustment_file()[0] != '\0') {
 104     const smpi_trace_call_location_t* loc                      = smpi_process()->call_location();
 105     std::string key                                            = loc->get_composed_key();
 106     std::unordered_map<std::string, double>::const_iterator it = location2speedup.find(key);
 107     if (it != location2speedup.end()) {
 108       speedup = it->second;
 109     }
 110   }
 111   return speedup;
 112 }
 113
 114 void smpi_bench_end()
 115 {
 116   if (MC_is_active() || MC_record_replay_is_active())
 117     return;
 118
 119   xbt_os_timer_t timer = smpi_process()->timer();
 120   xbt_os_threadtimer_stop(timer);
 121
 122 #if HAVE_PAPI
 123   /**
 124    * An MPI function has been called and now is the right time to update
 125    * our PAPI counters for this process.
 126    */
 127   if (not smpi_cfg_papi_events_file().empty()) {
 128     papi_counter_t& counter_data        = smpi_process()->papi_counters();
 129     int event_set                       = smpi_process()->papi_event_set();
 130     std::vector<long long> event_values(counter_data.size());
 131
 132     if (event_set)
 133       xbt_assert(PAPI_stop(event_set, &event_values[0]) == PAPI_OK, "Could not stop PAPI counters.");
 134     for (unsigned int i = 0; i < counter_data.size(); i++)
 135       counter_data[i].second += event_values[i];
 136   }
 137 #endif
 138
 139   if (smpi_process()->sampling()) {
 140     XBT_CRITICAL("Cannot do recursive benchmarks.");
 141     XBT_CRITICAL("Are you trying to make a call to MPI within an SMPI_SAMPLE_ block?");
 142     xbt_backtrace_display_current();
 143     xbt_die("Aborting.");
 144   }
 145
 146   // Maybe we need to artificially speed up or slow down our computation based on our statistical analysis.
 147   // Simulate the benchmarked computation unless disabled via command-line argument
 148   if (smpi_cfg_simulate_computation()) {
 149     smpi_execute(xbt_os_timer_elapsed(timer)/smpi_adjust_comp_speed());
 150   }
 151
 152 #if HAVE_PAPI
 153   if (not smpi_cfg_papi_events_file().empty() && TRACE_smpi_is_enabled()) {
 154     simgrid::instr::Container* container =
 155         simgrid::instr::Container::by_name(std::string("rank-") + std::to_string(simgrid::s4u::this_actor::get_pid()));
 156     const papi_counter_t& counter_data = smpi_process()->papi_counters();
 157
 158     for (auto const& pair : counter_data) {
 159       container->get_variable(pair.first)->set_event(SIMIX_get_clock(), pair.second);
 160     }
 161   }
 162 #endif
 163
 164   simgrid::smpi::utils::add_benched_time(xbt_os_timer_elapsed(timer));
 165 }
 166
 167 /* Private sleep function used by smpi_sleep(), smpi_usleep() and friends */
 168 static unsigned int private_sleep(double secs)
 169 {
 170   smpi_bench_end();
 171
 172   XBT_DEBUG("Sleep for: %lf secs", secs);
 173   aid_t pid = simgrid::s4u::this_actor::get_pid();
 174   TRACE_smpi_sleeping_in(pid, secs);
 175
 176   simgrid::s4u::this_actor::sleep_for(secs);
 177
 178   TRACE_smpi_sleeping_out(pid);
 179
 180   smpi_bench_begin();
 181   return 0;
 182 }
 183
 184 unsigned int smpi_sleep(unsigned int secs)
 185 {
 186   if (not smpi_process())
 187     return sleep(secs);
 188   return private_sleep(secs);
 189 }
 190
 191 int smpi_usleep(useconds_t usecs)
 192 {
 193   if (not smpi_process())
 194     return usleep(usecs);
 195   return static_cast<int>(private_sleep(usecs / 1000000.0));
 196 }
 197
 198 #if _POSIX_TIMERS > 0
 199 int smpi_nanosleep(const struct timespec* tp, struct timespec* t)
 200 {
 201   if (not smpi_process())
 202     return nanosleep(tp,t);
 203   return static_cast<int>(private_sleep(tp->tv_sec + tp->tv_nsec / 1000000000.0));
 204 }
 205 #endif
 206
 207 int smpi_gettimeofday(struct timeval* tv, struct timezone* tz)
 208 {
 209   if (not smpi_process()->initialized() || smpi_process()->finalized() || smpi_process()->sampling())
 210     return gettimeofday(tv, tz);
 211
 212   smpi_bench_end();
 213   double now = SIMIX_get_clock();
 214   if (tv) {
 215     tv->tv_sec = static_cast<time_t>(now);
 216 #ifdef WIN32
 217     tv->tv_usec = static_cast<useconds_t>((now - tv->tv_sec) * 1e6);
 218 #else
 219     tv->tv_usec = static_cast<suseconds_t>((now - tv->tv_sec) * 1e6);
 220 #endif
 221   }
 222   if (smpi_wtime_sleep > 0)
 223     simgrid::s4u::this_actor::sleep_for(smpi_wtime_sleep);
 224   smpi_bench_begin();
 225   return 0;
 226 }
 227
 228 #if _POSIX_TIMERS > 0
 229 int smpi_clock_gettime(clockid_t clk_id, struct timespec* tp)
 230 {
 231   if (not tp) {
 232     errno = EFAULT;
 233     return -1;
 234   }
 235   if (not smpi_process()->initialized() || smpi_process()->finalized() || smpi_process()->sampling())
 236     return clock_gettime(clk_id, tp);
 237   //there is only one time in SMPI, so clk_id is ignored.
 238   smpi_bench_end();
 239   double now = SIMIX_get_clock();
 240   tp->tv_sec  = static_cast<time_t>(now);
 241   tp->tv_nsec = static_cast<long int>((now - tp->tv_sec) * 1e9);
 242   if (smpi_wtime_sleep > 0)
 243     simgrid::s4u::this_actor::sleep_for(smpi_wtime_sleep);
 244   smpi_bench_begin();
 245   return 0;
 246 }
 247 #endif
 248
 249 double smpi_mpi_wtime()
 250 {
 251   double time;
 252   if (smpi_process()->initialized() && not smpi_process()->finalized() && not smpi_process()->sampling()) {
 253     smpi_bench_end();
 254     time = SIMIX_get_clock();
 255     if (smpi_wtime_sleep > 0)
 256       simgrid::s4u::this_actor::sleep_for(smpi_wtime_sleep);
 257     smpi_bench_begin();
 258   } else {
 259     time = SIMIX_get_clock();
 260   }
 261   return time;
 262 }
 263
 264 extern double sg_surf_precision;
 265 unsigned long long smpi_rastro_resolution ()
 266 {
 267   smpi_bench_end();
 268   double resolution = (1/sg_surf_precision);
 269   smpi_bench_begin();
 270   return static_cast<unsigned long long>(resolution);
 271 }
 272
 273 unsigned long long smpi_rastro_timestamp ()
 274 {
 275   smpi_bench_end();
 276   double now = SIMIX_get_clock();
 277
 278   auto sec               = static_cast<unsigned long long>(now);
 279   unsigned long long pre = (now - sec) * smpi_rastro_resolution();
 280   smpi_bench_begin();
 281   return sec * smpi_rastro_resolution() + pre;
 282 }
 283
 284 /* ****************************** Functions related to the SMPI_SAMPLE_ macros ************************************/
 285 namespace {
 286 class SampleLocation : public std::string {
 287 public:
 288   SampleLocation(bool global, const char* file, int line) : std::string(std::string(file) + ":" + std::to_string(line))
 289   {
 290     if (not global)
 291       this->append(":" + std::to_string(simgrid::s4u::this_actor::get_pid()));
 292   }
 293 };
 294
 295 class LocalData {
 296 public:
 297   double threshold; /* maximal stderr requested (if positive) */
 298   double relstderr; /* observed stderr so far */
 299   double mean;      /* mean of benched times, to be used if the block is disabled */
 300   double sum;       /* sum of benched times (to compute the mean and stderr) */
 301   double sum_pow2;  /* sum of the square of the benched times (to compute the stderr) */
 302   int iters;        /* amount of requested iterations */
 303   int count;        /* amount of iterations done so far */
 304   bool benching;    /* true: we are benchmarking; false: we have enough data, no bench anymore */
 305
 306   bool need_more_benchs() const;
 307 };
 308
 309 bool LocalData::need_more_benchs() const
 310 {
 311   bool res = (count < iters) || (threshold > 0.0 && (count < 2 ||          // not enough data
 312                                                      relstderr > threshold // stderr too high yet
 313                                                      ));
 314   XBT_DEBUG("%s (count:%d iter:%d stderr:%f thres:%f mean:%fs)",
 315             (res ? "need more data" : "enough benchs"), count, iters, relstderr, threshold, mean);
 316   return res;
 317 }
 318
 319 std::unordered_map<SampleLocation, LocalData, std::hash<std::string>> samples;
 320 }
 321
 322 void smpi_sample_1(int global, const char *file, int line, int iters, double threshold)
 323 {
 324   SampleLocation loc(global, file, line);
 325   if (not smpi_process()->sampling()) { /* Only at first call when benchmarking, skip for next ones */
 326     smpi_bench_end();     /* Take time from previous, unrelated computation into account */
 327     smpi_process()->set_sampling(1);
 328   }
 329
 330   auto insert = samples.emplace(loc, LocalData{
 331                                          threshold, // threshold
 332                                          0.0,       // relstderr
 333                                          0.0,       // mean
 334                                          0.0,       // sum
 335                                          0.0,       // sum_pow2
 336                                          iters,     // iters
 337                                          0,         // count
 338                                          true       // benching (if we have no data, we need at least one)
 339                                      });
 340   if (insert.second) {
 341     XBT_DEBUG("XXXXX First time ever on benched nest %s.", loc.c_str());
 342     xbt_assert(threshold > 0 || iters > 0,
 343         "You should provide either a positive amount of iterations to bench, or a positive maximal stderr (or both)");
 344   } else {
 345     LocalData& data = insert.first->second;
 346     if (data.iters != iters || data.threshold != threshold) {
 347       XBT_ERROR("Asked to bench block %s with different settings %d, %f is not %d, %f. "
 348                 "How did you manage to give two numbers at the same line??",
 349                 loc.c_str(), data.iters, data.threshold, iters, threshold);
 350       THROW_IMPOSSIBLE;
 351     }
 352
 353     // if we already have some data, check whether sample_2 should get one more bench or whether it should emulate
 354     // the computation instead
 355     data.benching = data.need_more_benchs();
 356     XBT_DEBUG("XXXX Re-entering the benched nest %s. %s", loc.c_str(),
 357               (data.benching ? "more benching needed" : "we have enough data, skip computes"));
 358   }
 359 }
 360
 361 int smpi_sample_2(int global, const char *file, int line, int iter_count)
 362 {
 363   SampleLocation loc(global, file, line);
 364
 365   XBT_DEBUG("sample2 %s %d", loc.c_str(), iter_count);
 366   auto sample = samples.find(loc);
 367   xbt_assert(sample != samples.end(),
 368              "Y U NO use SMPI_SAMPLE_* macros? Stop messing directly with smpi_sample_* functions!");
 369   const LocalData& data = sample->second;
 370
 371   if (data.benching) {
 372     // we need to run a new bench
 373     XBT_DEBUG("benchmarking: count:%d iter:%d stderr:%f thres:%f; mean:%f; total:%f",
 374               data.count, data.iters, data.relstderr, data.threshold, data.mean, data.sum);
 375     smpi_bench_begin();
 376   } else {
 377     // Enough data, no more bench (either we got enough data from previous visits to this benched nest, or we just
 378     //ran one bench and need to bail out now that our job is done). Just sleep instead
 379     if (not data.need_more_benchs()){
 380       XBT_DEBUG("No benchmark (either no need, or just ran one): count >= iter (%d >= %d) or stderr<thres (%f<=%f). "
 381               "Mean is %f, will be injected %d times",
 382               data.count, data.iters, data.relstderr, data.threshold, data.mean, iter_count);
 383
 384       //we ended benchmarking, let's inject all the time, now, and fast forward out of the loop.
 385       smpi_process()->set_sampling(0);
 386       smpi_execute(data.mean*iter_count);
 387       smpi_bench_begin();
 388       return 0;
 389     } else {
 390       XBT_DEBUG("Skipping - Benchmark already performed - accumulating time");
 391       xbt_os_threadtimer_start(smpi_process()->timer());
 392     }
 393   }
 394   return 1;
 395 }
 396
 397 void smpi_sample_3(int global, const char *file, int line)
 398 {
 399   SampleLocation loc(global, file, line);
 400
 401   XBT_DEBUG("sample3 %s", loc.c_str());
 402   auto sample = samples.find(loc);
 403   xbt_assert(sample != samples.end(),
 404              "Y U NO use SMPI_SAMPLE_* macros? Stop messing directly with smpi_sample_* functions!");
 405   LocalData& data = sample->second;
 406
 407   if (not data.benching)
 408     THROW_IMPOSSIBLE;
 409
 410   // ok, benchmarking this loop is over
 411   xbt_os_threadtimer_stop(smpi_process()->timer());
 412
 413   // update the stats
 414   data.count++;
 415   double period  = xbt_os_timer_elapsed(smpi_process()->timer());
 416   data.sum      += period;
 417   data.sum_pow2 += period * period;
 418   double n       = data.count;
 419   data.mean      = data.sum / n;
 420   data.relstderr = sqrt((data.sum_pow2 / n - data.mean * data.mean) / n) / data.mean;
 421
 422   XBT_DEBUG("Average mean after %d steps is %f, relative standard error is %f (sample was %f)",
 423             data.count, data.mean, data.relstderr, period);
 424
 425   // That's enough for now, prevent sample_2 to run the same code over and over
 426   data.benching = false;
 427 }
 428
 429 int smpi_sample_exit(int global, const char *file, int line, int iter_count){
 430   if (smpi_process()->sampling()){
 431     SampleLocation loc(global, file, line);
 432
 433     XBT_DEBUG("sample exit %s", loc.c_str());
 434     auto sample = samples.find(loc);
 435     xbt_assert(sample != samples.end(),
 436                "Y U NO use SMPI_SAMPLE_* macros? Stop messing directly with smpi_sample_* functions!");
 437
 438     if (smpi_process()->sampling()){//end of loop, but still sampling needed
 439       const LocalData& data = sample->second;
 440       smpi_process()->set_sampling(0);
 441       smpi_execute(data.mean * iter_count);
 442       smpi_bench_begin();
 443     }
 444   }
 445   return 0;
 446 }
 447
 448 smpi_trace_call_location_t* smpi_trace_get_call_location()
 449 {
 450   return smpi_process()->call_location();
 451 }
 452
 453 void smpi_trace_set_call_location(const char* file, const int line)
 454 {
 455   smpi_trace_call_location_t* loc = smpi_process()->call_location();
 456
 457   loc->previous_filename   = loc->filename;
 458   loc->previous_linenumber = loc->linenumber;
 459   if(not smpi_cfg_trace_call_use_absolute_path())
 460     loc->filename = simgrid::xbt::Path(file).get_base_name();
 461   else
 462     loc->filename = file;
 463   loc->linenumber = line;
 464 }
 465
 466 /** Required for Fortran bindings */
 467 void smpi_trace_set_call_location_(const char* file, const int* line)
 468 {
 469   smpi_trace_set_call_location(file, *line);
 470 }
 471
 472 /** Required for Fortran if -fsecond-underscore is activated */
 473 void smpi_trace_set_call_location__(const char* file, const int* line)
 474 {
 475   smpi_trace_set_call_location(file, *line);
 476 }
 477
 478 void smpi_bench_destroy()
 479 {
 480   samples.clear();
 481 }
 482
 483 int smpi_getopt_long_only (int argc,  char *const *argv,  const char *options,
 484                       const struct option * long_options, int *opt_index)
 485 {
 486   if (smpi_process())
 487     optind = smpi_process()->get_optind();
 488   int ret = getopt_long_only (argc,  argv,  options, long_options, opt_index);
 489   if (smpi_process())
 490     smpi_process()->set_optind(optind);
 491   return ret;
 492 }
 493
 494 int smpi_getopt_long (int argc,  char *const *argv,  const char *options,
 495                       const struct option * long_options, int *opt_index)
 496 {
 497   if (smpi_process())
 498     optind = smpi_process()->get_optind();
 499   int ret = getopt_long (argc,  argv,  options, long_options, opt_index);
 500   if (smpi_process())
 501     smpi_process()->set_optind(optind);
 502   return ret;
 503 }
 504
 505 int smpi_getopt (int argc,  char *const *argv,  const char *options)
 506 {
 507   if (smpi_process())
 508     optind = smpi_process()->get_optind();
 509   int ret = getopt (argc,  argv,  options);
 510   if (smpi_process())
 511     smpi_process()->set_optind(optind);
 512   return ret;
 513 }