XCheckAlgorithm.h

#pragma once
// cross check an algorithm running it with different ways
 
#include <iguana/algorithms/AlgorithmSequence.h>
#ifdef USE_HIPO_MULTITHREADED_CHAIN
#include "Fuzzing.h"
#include <hipo4/chain.h>
#include <hipo4/reader.h>
#include <unordered_set>
#endif
#include <unistd.h>
 
inline void FailXCheckAlgorithm() noexcept(false)
{
  throw std::runtime_error("Algorithm cross check test (xcheckalgo) failed");
}
 
inline void InvestigateBanks(hipo::banklist& banks, iguana::Logger& log)
{
#ifdef USE_HIPO_MULTITHREADED_CHAIN
  log.Error("Num banks: {}", banks.size());
  // redirect stdout fd to stderr fd, since `hipo::bank::show` uses `printf`
  int saved_stdout = dup(STDOUT_FILENO);
  dup2(STDERR_FILENO, STDOUT_FILENO);
  // show banks
  for(auto& bank : banks) {
    printf("Investigate bank '%s':\n", bank.getSchema().getName().c_str());
    printf("  - checksum 0x%zx\n", bank.checksum());
    printf("  - num rows filtered (total): %zu (%d)\n", bank.getRowList().size(), bank.getRows());
    bank.show();
  }
  // restore stderr redirection
  dup2(saved_stdout, STDOUT_FILENO);
  close(saved_stdout);
  log.Error("-----------------------------------------------------------------");
#endif
}
 
// NOTE: the cross check must process ALL events, otherwise the multi-threaded reader may
// sample a DIFFERENT set of events than the single-threaded reader
inline int XCheckAlgorithm(
    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 data_tag,
    int const num_threads,
    bool const vary_run,
    std::string const log_level,
    std::optional<int> investigate_evnum = std::nullopt)
{
 
#ifdef USE_HIPO_MULTITHREADED_CHAIN
 
  // set concurrency model; see `Fuzzing` constructor for why this is necessary
  iguana::GlobalConcurrencyModel = "memoize";
 
  // set logger for this test
  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 'xcheckalgo");
    return 1;
  }
 
  // put `RUN::config` bank up front
  hipo::banklist::size_type const run_config_bank_idx = 0;
  std::vector<std::string> bank_names_ordered;
  bank_names_ordered.push_back("RUN::config");
  for(auto const& name : bank_names) {
    if(name != "RUN::config")
      bank_names_ordered.push_back(name);
  }
 
  // setup run number variation
  Fuzzing fuzz;
 
  // mutex
  std::mutex mtx;
 
  // total number of events
  int num_events;
 
  // checksum data structure `xcheck_t`:
  // {
  //   { event1 number, { bank1 checksum, bank2 checksum, ... },
  //   { event2 number, { bank1 checksum, bank2 checksum, ... },
  //   ...,
  // }
  using evnum_t         = int;
  using checksum_list_t = std::vector<std::size_t>;
  using xcheck_elem_t   = std::pair<evnum_t, checksum_list_t>;
  using xcheck_t        = std::vector<xcheck_elem_t>;
 
  // ----------------------------------------------------------------------------------
  // run single threaded ("st")
  // ----------------------------------------------------------------------------------
 
  // xcheck data structures
  xcheck_t st_checksums;
  std::size_t st_num_banks;
  std::vector<std::string> st_bank_names;
 
  // test scope
  log.Info("Run single-threaded reader...");
  {
    // open the HIPO file
    hipo::reader r(data_file.c_str(), {data_tag});
    auto banks = r.getBanks(bank_names_ordered);
    num_events = r.getEntries();
 
    // define the algorithm
    iguana::AlgorithmSequence seq("st");
    for(auto const& prerequisite_algo : prerequisite_algos)
      seq.Add(prerequisite_algo);
    seq.Add(algo_name);
    seq.SetLogLevel(algo_name, log_level);
    seq.iguana::Algorithm::SetLogLevel(log_level);
 
    // start the algorithm
    seq.Start(banks);
 
    // get bank info
    st_num_banks = banks.size();
    for(auto& bank : banks)
      st_bank_names.push_back(bank.getSchema().getName());
 
    // event loop
    while(r.next(banks)) { // NOTE: must cross check ALL events
      // get event number
      auto& run_config_bank = banks[run_config_bank_idx];
      auto const evnum      = run_config_bank.getInt("event", 0);
      // run number variation
      if(vary_run) {
        auto random_run = fuzz.RunNumMod(evnum);
        if(random_run.has_value())
          run_config_bank.putInt("run", 0, random_run.value());
      }
      // run the algorithm
      seq.Run(banks);
      // bank checksums
      checksum_list_t checksums;
      for(auto const& bank : banks)
        checksums.push_back(bank.checksum());
      st_checksums.push_back({evnum, checksums});
      // investigate an event
      if(investigate_evnum && investigate_evnum.value() == evnum) {
        log.Error("Example single-threaded event {} that failed cross check:", evnum);
        InvestigateBanks(banks, log);
      }
    }
 
    // stop the algorithm
    seq.Stop();
  }
 
 
  // ----------------------------------------------------------------------------------
  // run multi threaded ("mt")
  // ----------------------------------------------------------------------------------
 
  // xcheck data structures
  xcheck_t mt_checksums;
  std::size_t mt_num_banks;
  std::vector<std::string> mt_bank_names;
 
  // test scope
  log.Info("Run multi-threaded reader...");
  {
 
    // start the HIPO chain
    hipo::chain ch(num_threads, false, false);
    ch.set_tags({data_tag});
    ch.add(data_file);
    auto banks = ch.getBanks(bank_names_ordered);
 
    // define the algorithm
    iguana::AlgorithmSequence seq("mt");
    for(auto const& prerequisite_algo : prerequisite_algos)
      seq.Add(prerequisite_algo);
    seq.Add(algo_name);
    seq.SetLogLevel(algo_name, log_level);
    seq.iguana::Algorithm::SetLogLevel(log_level);
 
    // start the algorithm
    seq.Start(banks);
 
    // get bank info
    mt_num_banks = banks.size();
    for(auto& bank : banks)
      mt_bank_names.push_back(bank.getSchema().getName());
 
    // define the event processor
    auto processor = [&](auto& banks, int file_idx, long event_idx) {
      // get event number
      auto& run_config_bank = banks[run_config_bank_idx];
      auto const evnum      = run_config_bank.getInt("event", 0);
      // run number variation
      if(vary_run) {
        auto random_run = fuzz.RunNumMod(evnum);
        if(random_run.has_value())
          run_config_bank.putInt("run", 0, random_run.value());
      }
      // run the iguana algorithm
      seq.Run(banks);
      // bank checksums
      checksum_list_t checksums;
      for(auto const& bank : banks)
        checksums.push_back(bank.checksum());
      std::lock_guard<std::mutex> const lock(mtx);
      mt_checksums.push_back({evnum, checksums});
      // investigate an event
      if(investigate_evnum && investigate_evnum.value() == evnum) {
        log.Error("Example multi-threaded event {} that failed cross check:", evnum);
        InvestigateBanks(banks, log);
      }
    };
 
    // process events
    ch.process(banks, processor, 100); // NOTE: must cross check ALL events
 
    // stop the algorithm
    seq.Stop();
  }
 
  // ----------------------------------------------------------------------------------
  // cross check
  // ----------------------------------------------------------------------------------
 
  // stop now, if investigating an event; return 0 since we are not cross checking
  // in this case
  if(investigate_evnum)
    return 0;
  // otherwise proceed with the cross check
  log.Info("Cross checking...");
 
  // xcheck the number of banks processed
  if(st_num_banks != mt_num_banks) {
    log.Error("readers processed different numbers of banks");
    FailXCheckAlgorithm();
  }
  auto const num_banks = mt_num_banks;
 
  // xcheck the bank names
  for(decltype(mt_bank_names)::size_type b = 0; b < mt_bank_names.size(); b++) {
    if(mt_bank_names.at(b) != st_bank_names.at(b)) {
      log.Error("readers processed different bank list");
      FailXCheckAlgorithm();
    }
  }
  auto bank_names_full = mt_bank_names;
 
  // checksum lists should have some events
  if(st_checksums.empty() || mt_checksums.empty()) {
    log.Error("no events were processed");
    FailXCheckAlgorithm();
  }
 
  // checksum lists must have the same size, otherwise they are definitely not equal
  if(st_checksums.size() != mt_checksums.size()) {
    log.Error("number of processed events differ");
    log.Error("  st_checksums.size() = {}", st_checksums.size());
    log.Error("  mt_checksums.size() = {}", mt_checksums.size());
    FailXCheckAlgorithm();
  }
 
  // sort checksum lists by event number
  auto sort_ftn = [](xcheck_elem_t const& a, xcheck_elem_t const& b) {
    if(a.second.empty() || b.second.empty())
      throw std::runtime_error("empty checksum vector");
    return a.first < b.first;
  };
  std::sort(st_checksums.begin(), st_checksums.end(), sort_ftn);
  std::sort(mt_checksums.begin(), mt_checksums.end(), sort_ftn);
 
  // check for any duplicate event numbers; they are very rare, and may come from EVIO files...
  auto dupe_check = [&log](xcheck_t const& xcheck, std::string id) {
    std::unordered_set<evnum_t> seen;
    std::vector<evnum_t> dupes;
    for(auto const& it : xcheck) {
      auto evnum = it.first;
      if(!seen.insert(evnum).second)
        dupes.push_back(evnum);
    }
    if(dupes.size() > 0) {
      std::sort(dupes.begin(), dupes.end());
      log.Warn("duplicate events found in {}:", id);
      log.Warn("  {{{}}}", fmt::join(dupes, ", "));
    }
    return dupes;
  };
  auto const st_dupes = dupe_check(st_checksums, "single-threaded reading");
  auto const mt_dupes = dupe_check(st_checksums, "multi-threaded reading");
  if(st_dupes != mt_dupes) {
    log.Error("list of duplicate events differs between readers");
    FailXCheckAlgorithm();
  }
  auto const& dupes = st_dupes;
  if((double)dupes.size() / num_events > 1e-4) {
    log.Error("there are too many duplicate events in these data");
    FailXCheckAlgorithm();
  }
 
  // cross check checksum lists
  unsigned int num_failed = 0;
  for(xcheck_t::size_type i = 0; i < mt_checksums.size(); i++) {
    bool fail_event          = false;
    auto const& st_ev        = st_checksums.at(i);
    auto const& mt_ev        = mt_checksums.at(i);
    auto const& st_evnum     = st_ev.first;
    auto const& mt_evnum     = mt_ev.first;
    auto const& st_checksums = st_ev.second;
    auto const& mt_checksums = mt_ev.second;
    // check event number
    if(st_evnum != mt_evnum) {
      log.Error("event number differs");
      log.Error("  single-threaded evnum: {}", st_evnum);
      log.Error("  multi-threaded evnum:  {}", mt_evnum);
      fail_event = true;
    }
    auto const& evnum = st_evnum;
    // check expected number of banks
    if(st_checksums.size() != num_banks) {
      log.Error("found event from single-threaded reader with unexpected number of banks");
      FailXCheckAlgorithm();
    }
    if(mt_checksums.size() != num_banks) {
      log.Error("found event from multi-threaded reader with unexpected number of banks");
      FailXCheckAlgorithm();
    }
    // skip duplicate events
    if(std::find(dupes.begin(), dupes.end(), evnum) != dupes.end()) {
      log.Warn("not crosschecking duplicate event {}", evnum);
      continue;
    }
    // loop over banks' checksums, cross checking each
    for(std::size_t b = 0; b < num_banks; b++) {
      if(st_checksums.at(b) != mt_checksums.at(b)) {
        log.Error("xcheck fail: bank {:?} differs", bank_names_full.at(b));
        log.Error("  event number:             {}", evnum);
        log.Error("  single-threaded checksum: 0x{:x}", st_checksums.at(b));
        log.Error("  multi-threaded checksum:  0x{:x}", mt_checksums.at(b));
        fail_event = true;
      }
    }
    // investigate the first failed event: re-run `XCheckAlgorithm` and show banks for this event
    if(fail_event && num_failed == 0) {
      log.Error("=========== showing all banks for this failure ===========");
      auto r = XCheckAlgorithm(
          algo_name,
          prerequisite_algos,
          bank_names,
          data_file,
          data_tag,
          num_threads,
          vary_run,
          log_level,
          st_evnum);
      if(r != 0)
        return r; // return nonzero if the investigation failed
      log.Error("=========== proceeding with cross check of remaining events ===========");
    }
    // fail early if too many events fail
    if(fail_event) {
      num_failed++;
      if(num_failed > 100) {
        log.Error("too many events failed, aborting cross check now");
        FailXCheckAlgorithm();
      }
    }
  }
  // fail if any events failed
  if(num_failed > 0) {
    log.Error("{} events failed cross check", num_failed);
    FailXCheckAlgorithm();
  }
 
  // ----------------------------------------------------------------------------------
  log.Info("Cross check success");
  return 0;
 
#else
  throw std::runtime_error("called `xcheckalgo` but HIPO version does not support it");
#endif
}