TestMultithreading.h

// multithreaded test of an iguana algorithm
 
#include <hipo4/reader.h>
#include <iguana/algorithms/AlgorithmSequence.h>
#include <random>
 
inline int TestMultithreading(
    std::string const command,
    std::string const algo_name,
    std::vector<std::string> const prerequisite_algos,
    std::vector<std::string> const bank_names,
    std::string const data_file,
    int const num_events,
    int const num_threads,
    std::string const concurrency_model,
    bool const vary_run,
    std::string const log_level)
{
 
  iguana::Logger log("test");
  log.SetLevel(log_level);
 
  // check arguments
  if(algo_name.empty() || bank_names.empty()) {
    log.Error("need algorithm name and banks");
    return 1;
  }
  if(data_file.empty()) {
    log.Error("need a data file for command {:?}", command);
    return 1;
  }
 
  // set the concurrency model
  if(!concurrency_model.empty())
    iguana::GlobalConcurrencyModel = concurrency_model;
 
  // find the 'RUN::config' bank, if any
  std::optional<hipo::banklist::size_type> run_config_bank_idx{};
  if(vary_run) {
    for(hipo::banklist::size_type idx = 0; idx < bank_names.size(); idx++) {
      if(bank_names.at(idx) == "RUN::config") {
        run_config_bank_idx = idx;
        break;
      }
    }
  }
 
  // number of events per thread
  int const default_frame_size = 50;
  int num_events_per_thread    = (int)std::round((double)num_events / num_threads);
  int num_events_per_frame     = num_events > 0 ? std::min(num_events_per_thread, default_frame_size) : default_frame_size;
  int num_frames_per_thread    = num_events > 0 ? (int)std::ceil((double)num_events_per_thread / num_events_per_frame) : 0;
  int num_events_actual        = num_events_per_frame * num_frames_per_thread * num_threads;
  log.Info("num_events_per_thread = {}", num_events_per_thread);
  log.Info("num_events_per_frame  = {}", num_events_per_frame);
  log.Info("num_frames_per_thread = {}", num_frames_per_thread);
  if(num_events > 0) {
    log.Info("=> will actually process num_events = {}", num_events_actual);
    if(num_events != num_events_actual)
      log.Warn("argument's num_events ({}) differs from the actual num_events that will be processed ({})",
               num_events, num_events_actual);
  }
  else {
    log.Info("=> will actually process num_events = ALL OF THEM");
  }
 
  // start the stream
  hipo::readerstream stream;
  stream.open(data_file.c_str());
 
  // define the worker function
  auto ftn = [&stream,
              algo_name,
              prerequisite_algos,
              bank_names,
              vary_run,
              log_level,
              num_events_per_thread,
              num_events_per_frame,
              run_config_bank_idx](int order) {
    // fill a frame
    std::vector<hipo::event> events;
    for(int i = 0; i < num_events_per_frame; i++)
      events.push_back(hipo::event());
 
    // bank list
    hipo::banklist banks;
    for(auto const& bank_name : bank_names)
      banks.push_back(hipo::bank(stream.dictionary().getSchema(bank_name.c_str()), 48));
 
    // define the algorithm
    iguana::AlgorithmSequence seq;
    for(auto const& prerequisite_algo : prerequisite_algos)
      seq.Add(prerequisite_algo);
    seq.Add(algo_name);
    seq.SetName("TEST thread " + std::to_string(order));
    seq.PrintSequence();
    seq.SetLogLevel(algo_name, log_level);
    seq.iguana::Algorithm::SetLogLevel(log_level);
 
    // start the algorithm
    seq.Start(banks);
 
    // setup run number variation
    std::vector<int> golden_runs = {
        // FIXME: for now, just RG-A, since some algorithms can _only_ run on RG-A
        4013, // rga_spring2018
        5038, // rga_fall2018
        // 5700, // rgk_fall2018
        // 6302, // rgb_spring2019
        6666, // rga_spring2019
        // 11234, // rgb_fall2019
        // 11567, // rgb_spring2020
        // 12933, // rgf_summer2020
        // 15833, // rgm_fall2021
        // 16600, // rgc_spring2022
        // 17407, // rgc_fall2022
        // 17800, // rgc_spring2023
        // 18779, // rgd_fall2023
        // 19877, // rgk_2024
        // 20522, // rge_2024
        // 23050, // rgl_2025
    };
    std::mt19937 rng(std::random_device{}());
    std::uniform_int_distribution<std::size_t> random_num(0, 9);
    std::uniform_int_distribution<std::size_t> random_golden_run(0, golden_runs.size() - 1);
    if(vary_run) {
      seq.GetLog()->Warn("==================================================================================================");
      seq.GetLog()->Warn("RUN NUMBER VARIATION IS ENABLED, DO NOT ATTEMPT TO CROSS CHECK RESULTS WITH SINGLE-THREADED TESTS!");
      seq.GetLog()->Warn("==================================================================================================");
    }
 
    // loop over frames
    int nProcessed = 0;
    while(nProcessed < num_events_per_thread || num_events_per_thread == 0) {
      stream.pull(events);
 
      // loop over events in this frame
      int nNonEmpty = 0;
      for(auto& event : events) {
        if(event.getSize() > 16) {
          nNonEmpty++;
          nProcessed++;
 
          // read the banks
          for(auto& bank : banks)
            event.read(bank);
 
          // occasionally vary the run number; so far, algorithms with data-dependent configuration
          // parameters have dependence on run number, so this variation aims to improve thread
          // sanitizer test coverage
          if(vary_run && run_config_bank_idx.has_value()) {
            if(random_num(rng) < 3) {
              auto runnum = golden_runs[random_golden_run(rng)];
              seq.GetLog()->Trace("RANDOMLY VARIED RUN NUMBER TO BE {}", runnum);
              banks[run_config_bank_idx.value()].putInt("run", 0, runnum);
            }
          }
 
          // run the iguana algorithm
          seq.Run(banks);
        }
      }
      if(nNonEmpty == 0)
        break;
    }
 
    // stop the algorithm
    seq.Stop();
    if(vary_run) {
      seq.GetLog()->Warn("=========================================================================================================");
      seq.GetLog()->Warn("RANDOM RUN NUMBER VARIATION IS ENABLED, DO NOT ATTEMPT TO CROSS CHECK RESULTS WITH SINGLE-THREADED TESTS!");
      seq.GetLog()->Warn("=========================================================================================================");
    }
 
    seq.GetLog()->Info("nProcessed = {}", nProcessed);
    return nProcessed;
  };
 
  // run
  stream.run(ftn, num_threads);
  return 0;
}