PrunDriver.cxx File Reference

#include <EventLoopGrid/PrunDriver.h>
#include <EventLoop/Algorithm.h>
#include <EventLoop/ManagerData.h>
#include <EventLoop/ManagerStep.h>
#include <EventLoop/Job.h>
#include <EventLoop/MessageCheck.h>
#include <EventLoop/OutputStream.h>
#include <PathResolver/PathResolver.h>
#include <RootCoreUtils/Assert.h>
#include <RootCoreUtils/hadd.h>
#include <RootCoreUtils/ShellExec.h>
#include <SampleHandler/MetaObject.h>
#include <SampleHandler/Sample.h>
#include <SampleHandler/SampleGrid.h>
#include <SampleHandler/SampleHandler.h>
#include <SampleHandler/GridTools.h>
#include <TList.h>
#include <TPython.h>
#include <TROOT.h>
#include <TFile.h>
#include <TSystem.h>
#include <algorithm>
#include <any>
#include <array>
#include <cstdlib>
#include <filesystem>
#include <iostream>
#include <optional>
#include <set>
#include <sstream>
#include <string>
#include <vector>
#include <stdexcept>
#include <ranges>
#include "pool.h"
#include <mutex>

Go to the source code of this file.

Functions
	ClassImp (EL::PrunDriver) namespace
static JobState::Enum	sampleState (SH::Sample *sample)
static JobState::Enum	nextState (JobState::Enum state, Status::Enum status)
static SH::MetaObject	defaultOpts ()
static std::mutex &	logMutex ()
static bool	downloadContainer (const std::string &name, const std::string &location)
static int	callPythonOnSample (const char *macroFile, const char *func, SH::Sample *sample)
static Status::Enum	submit (SH::Sample *const sample, const bool isFirstSample)
static Status::Enum	checkPandaTask (SH::Sample *const sample)
static Status::Enum	download (SH::Sample *const sample)
static Status::Enum	merge (SH::Sample *const sample)
static void	processTask (SH::Sample *const sample, const bool isFirstSample)
static void	processAllInState (const SH::SampleHandler &sh, JobState::Enum state, const size_t nThreads)
static std::optional< std::string >	gridNickname ()
static std::string	formatOutputName (const SH::MetaObject &sampleMeta, const std::string &pattern)
static std::string	outputFileNames (const EL::Job &job)
static void	saveJobDef (const std::string &fileName, const EL::Job &job, const SH::SampleHandler &sh)
static SH::SampleHandler	outputSH (const SH::SampleHandler &in, const std::string &outputLabel)

Function Documentation

callPythonOnSample()

int callPythonOnSample ( const char * macroFile, const char * func, SH::Sample * sample )

static

Definition at line 203 of file PrunDriver.cxx.

{
  static std::mutex mutex;
  std::lock_guard<std::mutex> lock(mutex);
 
  static std::set<std::string> loadedMacros;
  if (loadedMacros.insert(macroFile).second) {
    TPython::LoadMacro(PathResolverFindCalibFile(macroFile).c_str());
  }
 
  TPython::Bind(sample, "ELG_SAMPLE");
  std::any result;
  const std::string code =
    std::string("_anyresult = ROOT.std.make_any['int'](") + func + "(ELG_SAMPLE))";
  TPython::Exec(code.c_str(), &result);
  TPython::Bind(nullptr, "ELG_SAMPLE");
  return std::any_cast<int>(result);
}

checkPandaTask()

Status::Enum checkPandaTask ( SH::Sample *const sample )

static

Definition at line 256 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
  RCU_REQUIRE(static_cast<int>(sample->meta()->castDouble("nc_jediTaskID",0, SH::MetaObject::CAST_NOCAST_DEFAULT)) > 100);
 
  int ret = callPythonOnSample("EventLoopGrid/ELG_jediState.py", "ELG_jediState", sample);
 
  if (ret == Status::DONE) return Status::DONE;
  if (ret == Status::FAIL) return Status::FAIL;
 
  // Value 90 corresponds to `running` state of the job
  if (ret != 90) { sample->meta()->setString("nc_ELG_state_details", "task status other than done/finished/failed/running"); }
  // Value 99 is returned if there is error in the script (import, missing ID)
  if (ret == 99) {
    sample->meta()->setString("nc_ELG_state_details",
                              "problem checking jedi task status");
  }
 
  return Status::PENDING;
}

ClassImp()

ClassImp

(

EL::PrunDriver

)

Author

Alexander Madsen

Nils Krumnack

Definition at line 52 of file PrunDriver.cxx.

          {
  namespace JobState {  
    static const unsigned int NSTATES = 6;
    enum Enum { INIT=0, RUN=1, DOWNLOAD=2, MERGE=3, FINISHED=4, FAILED=5 };
    static const char* name[NSTATES] = 
      { "INIT", "RUNNING", "DOWNLOAD", "MERGE", "FINISHED", "FAILED" };
    Enum parse(const std::string& what)
    {
      for (unsigned int i = 0; i != NSTATES; ++i) {
    if (what == name[i]) { return static_cast<Enum>(i); }
      }
      throw std::runtime_error("PrunDriver.cxx: Failed to parse job state string");
    }
  }
 
  // When changing the values in the enum make sure
  // corresponding values in `data/ELG_jediState.py` script
  // are changed accordingly
  namespace Status {
    enum Enum { DONE=0, PENDING=1, FAIL=2 };
  }
 
  struct TransitionRule {
    JobState::Enum fromState;
    Status::Enum status;
    JobState::Enum toState;
  };
 
  struct TmpCd {
    const std::string origDir;
    TmpCd(const std::string & dir)
      : origDir(gSystem->pwd())
    {
      gSystem->cd(dir.c_str());
    }
    ~TmpCd()
    {
      gSystem->cd(origDir.c_str());
    }
  };
}

defaultOpts()

SH::MetaObject defaultOpts ( )

static

Definition at line 131 of file PrunDriver.cxx.

{
  SH::MetaObject o;
  o.setString("nc_nGBPerJob", "MAX");
  o.setString("nc_mergeOutput", "true");
  o.setString("nc_cmtConfig", gSystem->ExpandPathName("$AnalysisBase_PLATFORM"));
  o.setString("nc_useAthenaPackages", "true");
  const std::string mergestr = "elg_merge jobdef.root %OUT %IN";
  o.setString("nc_mergeScript", mergestr);
  return o;
}

download()

Status::Enum download ( SH::Sample *const sample )

static

Definition at line 277 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
  using namespace EL::msgEventLoop;
 
  // This can run on several download threads at once; the shared message stream
  // is not thread-safe, so serialize the logging (see logMutex()).
  {
    std::lock_guard<std::mutex> lock(logMutex());
    ANA_MSG_INFO("Downloading output from: " << sample->name() << "...");
  }
 
  std::string container = sample->meta()->castString("nc_outDS", "", SH::MetaObject::CAST_NOCAST_DEFAULT);
  RCU_ASSERT(not container.empty());
  if (container[container.size()-1] == '/') {
    container.resize(container.size() - 1);
  }
  container += "_hist/";
 
  bool downloadOk = downloadContainer(container, "elg/download/" + container);
 
  if (not downloadOk) {
    std::lock_guard<std::mutex> lock(logMutex());
    ANA_MSG_ERROR("Failed to download one or more files");
    sample->meta()->setString("nc_ELG_state_details",
                              "error, check log for details");
    return Status::PENDING;
  }
 
  return Status::DONE;
}

downloadContainer()

bool downloadContainer ( const std::string & name, const std::string & location )

static

Definition at line 154 of file PrunDriver.cxx.

{
  using namespace EL::msgEventLoop;
  RCU_ASSERT(not name.empty());
  RCU_ASSERT(name[name.size()-1] == '/');
  RCU_ASSERT(not location.empty());
 
  try {
    std::error_code ec;
    std::filesystem::create_directories(location, ec);
    if (ec) {
      std::lock_guard<std::mutex> lock(logMutex());
      ANA_MSG_ERROR("Failed to create directory " << location << ": " << ec.message());
      return false;
    }
 
    std::vector<std::string> datasets;
    for (auto& entry : SH::rucioListDids (name))
    {
      if (entry.type == "CONTAINER" || entry.type == "DIDType.CONTAINER")
        datasets.push_back (entry.name);
    }
 
    auto downloadResult = SH::rucioDownloadList (location, datasets);
    for (const auto& result : downloadResult)
    {
      if (result.notDownloaded != 0)
        return false;
    }
  } catch (const std::exception& e) {
    std::lock_guard<std::mutex> lock(logMutex());
    ANA_MSG_ERROR("Failed to download " << name << ": " << e.what());
    return false;
  }
  return true;
}

formatOutputName()

std::string formatOutputName ( const SH::MetaObject & sampleMeta, const std::string & pattern )

static

Definition at line 490 of file PrunDriver.cxx.

{
  const std::string sampleName = sampleMeta.castString("sample_name");
  RCU_REQUIRE(not pattern.empty());
  using namespace EL::msgEventLoop;
 
  TString out = pattern.c_str();
 
  // Handle case of no proxy; will create a proxy later in the submission
  const std::optional<std::string> nickname = gridNickname();
  if (not nickname.has_value()){
    ANA_MSG_WARNING( "No proxy available - cannot use nickname yet. Will try a late replacement.");
  } else {
    out.ReplaceAll("%nickname%", *nickname);
  }
 
  out.ReplaceAll("%in:name%", sampleName);
 
  std::stringstream ss(sampleName);
  std::string item;
  int field = 0;
  while(std::getline(ss, item, '.')) {
    std::stringstream sskey;
    sskey << "%in:name[" << ++field << "]%";
    out.ReplaceAll(sskey.str(), item);
  }
  while (out.Index("%in:") != -1) {
    int i1 = out.Index("%in:");
    int i2 = out.Index("%", i1+1);
    if (i2 == -1) {
      ANA_MSG_ERROR("malformed output name pattern, unterminated %in: token in \""
            << out.Data() << "\"");
      break;
    }
    TString metaName = out(i1+4, i2-i1-4);
    out.ReplaceAll("%in:"+metaName+"%",
           sampleMeta.castString(std::string(metaName.Data())));
  }
  out.ReplaceAll("/", "");
  return out.Data();
}

gridNickname()

std::optional< std::string > gridNickname ( )

static

Definition at line 467 of file PrunDriver.cxx.

{
  static std::optional<std::string> cached;
  if (cached.has_value()) { return cached; }
 
  std::any result;
  const char* code =
    "try:\n"
    "    from pandatools import PsubUtils\n"
    "    _nick = str(PsubUtils.getNickname())\n"
    "except Exception:\n"
    "    _nick = ''\n"
    "_anyresult = ROOT.std.make_any['std::string'](_nick)\n";
  TPython::Exec(code, &result);
  const std::string nickname = std::any_cast<std::string>(result);
 
  // An empty result means the lookup failed; a very long one is panda's
  // "no proxy" message rather than an actual nickname.
  if (nickname.empty() || nickname.length() > 20) { return std::nullopt; }
  cached = nickname;
  return cached;
}

logMutex()

std::mutex & logMutex ( )

static

Definition at line 148 of file PrunDriver.cxx.

{
  static std::mutex mutex;
  return mutex;
}

merge()

Status::Enum merge ( SH::Sample *const sample )

static

Definition at line 309 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
  // The MERGE state is always processed single-threaded, so the logging here
  // needs no serialization (unlike download(), see logMutex()).
  using namespace EL::msgEventLoop;
 
  std::string container = sample->meta()->castString("nc_outDS", "", SH::MetaObject::CAST_NOCAST_DEFAULT);
  RCU_ASSERT(not container.empty());
  if (container[container.size()-1] == '/') {
    container.resize(container.size() - 1);
  }
  container += "_hist/";
  const std::string dir = "elg/download/" + container;
 
  const std::string fileName = "hist-output.root";
 
  const std::string target = Form("hist-%s.root", sample->name().c_str());
 
  // Collect the per-job output files, i.e. those whose name contains
  // ".hist-output.root" (matching the old `find -name "*.hist-output.root*"`).
  namespace fs = std::filesystem;
  const std::string needle = "." + fileName;
  std::vector<std::string> files;
  std::error_code ec;
  for (fs::recursive_directory_iterator it(dir, ec), end; it != end; it.increment(ec)) {
    if (ec) { break; }
    if (it->is_regular_file() &&
        it->path().filename().string().find(needle) != std::string::npos) {
      files.push_back(it->path().string());
    }
  }
 
  std::sort(files.begin(), files.end());
  // Duplicates are not expected (the directory scan never lists a file twice),
  // but drop them with a real runtime check rather than a side-effecting assert.
  std::vector<std::string> duplicates;
  for (size_t i = 1; i < files.size(); ++i) {
    if (files[i] == files[i - 1]) { duplicates.push_back(files[i]); }
  }
  if (not duplicates.empty()) {
    std::ostringstream dup;
    for (const std::string& name : duplicates) { dup << ' ' << name; }
    ANA_MSG_WARNING("Ignoring duplicate input file(s) for merging:" << dup.str());
    files.erase(std::unique(files.begin(), files.end()), files.end());
  }
 
  if (not files.size()) {
    ANA_MSG_ERROR("Found no input files for merging! "
          "Requeueing sample for download...");
    sample->meta()->setString("nc_ELG_state_details", "retry, files were lost");
    return Status::FAIL;
  }
 
  try {
    RCU::hadd(target.c_str(), files);
  } catch (...) {
    sample->meta()->setString("nc_ELG_state_details",
                              "error, check log for details");
    fs::remove(target, ec);
    return Status::PENDING;
  }
 
  // Remove the merged inputs and the (now empty) download tree.
  for (const std::string& file : files) {
    fs::remove(file, ec);
    if (ec) {
      ANA_MSG_WARNING("Failed to remove merged input " << file << ": " << ec.message());
    }
  }
  fs::remove_all(dir, ec);
  if (ec) {
    ANA_MSG_WARNING("Failed to remove download directory " << dir << ": " << ec.message());
  }
 
  return Status::DONE;
}

nextState()

JobState::Enum nextState ( JobState::Enum state, Status::Enum status )

static

Definition at line 104 of file PrunDriver.cxx.

{
  RCU_REQUIRE(state != JobState::FINISHED);
  RCU_REQUIRE(state != JobState::FAILED);
  static constexpr std::array<TransitionRule, 12> TABLE =
    {{
      {JobState::INIT,     Status::DONE,    JobState::RUN},
      {JobState::INIT,     Status::PENDING, JobState::INIT},
      {JobState::INIT,     Status::FAIL,    JobState::FAILED},
      {JobState::RUN,      Status::DONE,    JobState::DOWNLOAD},
      {JobState::RUN,      Status::PENDING, JobState::RUN},
      {JobState::RUN,      Status::FAIL,    JobState::FAILED},
      {JobState::DOWNLOAD, Status::DONE,    JobState::MERGE},
      {JobState::DOWNLOAD, Status::PENDING, JobState::DOWNLOAD},
      {JobState::DOWNLOAD, Status::FAIL,    JobState::FAILED},
      {JobState::MERGE,    Status::DONE,    JobState::FINISHED},
      {JobState::MERGE,    Status::PENDING, JobState::MERGE},
      {JobState::MERGE,    Status::FAIL,    JobState::DOWNLOAD}
    }};
  for (const TransitionRule& rule : TABLE) {
    if (rule.fromState == state && rule.status == status) {
      return rule.toState;
    }
  }
  throw std::logic_error("PrunDriver.cxx: Missing state transition rule");
}

outputFileNames()

std::string outputFileNames ( const EL::Job & job )

static

Definition at line 533 of file PrunDriver.cxx.

{
  std::string out = "hist:hist-output.root";
  for (EL::Job::outputIter os = job.outputBegin(),
     end = job.outputEnd(); os != end; ++os) {
    const std::string name = os->label() + ".root";
    const std::string ds =
      os->options()->castString(EL::OutputStream::optContainerSuffix);
    out += "," + (ds.empty() ? name : ds + ":" + name);
  }
  return out;
}

outputSH()

SH::SampleHandler outputSH ( const SH::SampleHandler & in, const std::string & outputLabel )

static

Definition at line 571 of file PrunDriver.cxx.

{
  SH::SampleHandler out;
  const std::string outputFile = "*" + outputLabel + ".root*";
  const std::string outDSSuffix = '_' + outputLabel + ".root/"; 
  for (SH::Sample* const sample : in) {
    auto outSample = std::make_unique<SH::SampleGrid>(sample->name());
    const std::string outputDS = sample->meta()->castString("nc_outDS", "", SH::MetaObject::CAST_NOCAST_DEFAULT) + outDSSuffix;
    outSample->meta()->setString("nc_grid", outputDS);
    outSample->meta()->setString("nc_grid_filter", outputFile);
    out.add(std::move(outSample));
  }
  out.fetch(in);
  return out;
}

processAllInState()

void processAllInState ( const SH::SampleHandler & sh, JobState::Enum state, const size_t nThreads )

static

Definition at line 418 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sh.size());
 
  WorkList workList;
 
  bool isFirstSample = true;
  for (SH::Sample* const sample : sh) {
    if (sampleState(sample) == state) {
      workList.push_back([sample, isFirstSample, state]()->void{
        if (state == JobState::INIT) {
          // INIT is always processed single-threaded, so let an exception
          // (e.g. the deliberate tarball-creation abort in submit()) propagate
          // and stop the submission.
          processTask(sample, isFirstSample);
        } else {
          // On a worker thread an escaping exception would call std::terminate.
          // Record it against the sample and mark it FAILED so the pool
          // finishes processing the remaining samples.
          try {
            processTask(sample, isFirstSample);
          } catch (const std::exception& e) {
            using namespace EL::msgEventLoop;
            {
              std::lock_guard<std::mutex> lock(logMutex());
              ANA_MSG_ERROR ("Exception while processing " << sample->name()
                             << ": " << e.what());
            }
            sample->meta()->setString ("nc_ELG_state_details",
                                       std::string ("exception: ") + e.what());
            sample->meta()->setString ("nc_ELG_state",
                                       JobState::name[JobState::FAILED]);
          }
        }
      });
      // Change boolean to false as already processed one sample
      isFirstSample = false;
    }
  }
  process(workList, nThreads);
}

processTask()

void processTask ( SH::Sample *const sample, const bool isFirstSample )

static

Definition at line 387 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
 
  JobState::Enum state = sampleState(sample);
  
  sample->meta()->setString("nc_ELG_state_details", "");
 
  Status::Enum status = Status::PENDING;
  switch (state) {
  case JobState::INIT: 
    status = submit(sample, isFirstSample);
    break;
  case JobState::RUN: 
    status = checkPandaTask(sample);
    break;
  case JobState::DOWNLOAD: 
    status = download(sample);
    break;
  case JobState::MERGE: 
    status = merge(sample);
    break;
  case JobState::FINISHED: 
  case JobState::FAILED:
    break;
  }
 
  state = nextState(state, status);
  sample->meta()->setString("nc_ELG_state", JobState::name[state]);
}

sampleState()

JobState::Enum sampleState ( SH::Sample * sample )

static

Definition at line 96 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
  static const std::string defaultState = JobState::name[JobState::INIT];
  std::string label = sample->meta()->castString ("nc_ELG_state", defaultState, SH::MetaObject::CAST_NOCAST_DEFAULT);
  return JobState::parse(label);
}

saveJobDef()

void saveJobDef ( const std::string & fileName, const EL::Job & job, const SH::SampleHandler & sh )

static

Definition at line 547 of file PrunDriver.cxx.

{
  TFile file(fileName.c_str(), "RECREATE");
  TList outputs;
  outputs.SetOwner(true);
  for (EL::Job::outputIter o = job.outputBegin(); o !=job.outputEnd(); ++o)
    outputs.Add(o->Clone());
  file.WriteTObject(&job.jobConfig(), "jobConfig", "SingleKey");        
  file.WriteTObject(&outputs, "outputs", "SingleKey");        
  bool haveDefault = false;
  for (SH::Sample* const sample : sh) {
    const SH::MetaObject& meta = *(sample->meta());
    file.WriteObject(&meta, meta.castString("sample_name").c_str());
    if (!haveDefault)
    {
      file.WriteObject (&meta, "defaultMetaObject");
      haveDefault = true;
    }
  }
}  

submit()

Status::Enum submit ( SH::Sample *const sample, const bool isFirstSample )

static

Definition at line 223 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
  using namespace EL::msgEventLoop;
 
  ANA_MSG_INFO( "Submitting " << sample->name() << "..." );
 
  int ret = callPythonOnSample("EventLoopGrid/ELG_prun.py", "ELG_prun", sample);
 
  // Tarball is created for the first sample to be submitted
  // then the tarball is simply reused for the other samples 
  // If the returned value is 1 it implies the tarball creation failed 
  // See EventLoopGrid/data/ELG_prun.py script
  // Abort any further processing as the tarball was not succesfully created
  if (isFirstSample && ret == 1){
    ANA_MSG_ERROR("Failed to create tarball");
    throw std::runtime_error("PrunDriver.cxx: aborting due to tarball creation issue"); 
  }
 
  if (ret < 100) {
    sample->meta()->setString("nc_ELG_state_details", 
                              "problem submitting"); 
    return Status::FAIL;
  }
 
  sample->meta()->setDouble("nc_jediTaskID", ret);
 
  // Let's also tell people about their task ID
  ANA_MSG_INFO( "Task submitted; jediTaskID=" << ret );
 
  return Status::DONE;
}