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 <cstdlib>
#include <iostream>
#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 bool	downloadContainer (const std::string &name, const std::string &location)
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::string	formatOutputName (const SH::MetaObject &sampleMeta, const std::string &pattern)
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

checkPandaTask()

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

static

Definition at line 230 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
  RCU_REQUIRE(static_cast<int>(sample->meta()->castDouble("nc_jediTaskID",0, SH::MetaObject::CAST_NOCAST_DEFAULT)) > 100);
 
  static bool loaded = false;
  if (not loaded) {
    std::string path = PathResolverFindCalibFile("EventLoopGrid/ELG_jediState.py");
    TPython::LoadMacro(path.c_str());
    loaded = true;
  }
 
  TPython::Bind(dynamic_cast<TObject*>(sample), "ELG_SAMPLE");
#if ROOT_VERSION_CODE >= ROOT_VERSION(6,33,01)
  std::any result;
  TPython::Exec("_anyresult = ROOT.std.make_any['int'](ELG_jediState(ELG_SAMPLE))", &result);
  int ret = std::any_cast<int>(result);
#else
  int ret =  TPython::Eval("ELG_jediState(ELG_SAMPLE)");
#endif
  TPython::Bind(0, "ELG_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 47 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); }
      }
      RCU_ASSERT0("Failed to parse job state string");
      throw std::runtime_error("PrunDriver.cxx: Failed to parse job state string"); //compiler dummy
    }
  }
 
  // When changing the values in the enum make sure
  // corresponding values in `data/ELG_jediState.py` script
  // are changed accordingly
  namespace Status {
    //static const int NSTATES = 3;
    enum Enum { DONE=0, PENDING=1, FAIL=2 };
  }
 
  struct TransitionRule {
    JobState::Enum fromState; 
    Status::Enum status; 
    JobState::Enum toState; 
    TransitionRule(JobState::Enum fromState,
           Status::Enum status,
           JobState::Enum toState)
      : fromState(fromState)
      , status(status)
      , toState(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 138 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 266 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
 
  {
    static std::mutex mutex;
    std::lock_guard<std::mutex> lock(mutex);
    std::cout << "Downloading output from: " 
          << sample->name() << "..." << std::endl;
  }
 
  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::cerr << "Failed to download one or more files" << std::endl;
    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 150 of file PrunDriver.cxx.

{
  RCU_ASSERT(not name.empty()); 
  RCU_ASSERT(name[name.size()-1] == '/'); 
  RCU_ASSERT(not location.empty()); 
 
  try {
    gSystem->Exec(Form("mkdir -p %s", location.c_str()));
 
    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 (...) { 
    return false;
  }
  return true;
}

formatOutputName()

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

static

Definition at line 395 of file PrunDriver.cxx.

{
  const std::string sampleName = sampleMeta.castString("sample_name");
  RCU_REQUIRE(not pattern.empty());
  using namespace EL::msgEventLoop;
 
  static const std::string nickname = 
    gSystem->GetFromPipe(Form("python -c \"%s\" 2>/dev/null", 
                  "from pandatools import PsubUtils;"
                  "print(PsubUtils.getNickname());")).Data();
 
  TString out = pattern.c_str();
 
  // Handle case of no proxy; will create a proxy later in the submission
  if (nickname.length()>20){
    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);
    TString metaName = out(i1+4, i2-i1-4);
    out.ReplaceAll("%in:"+metaName+"%", 
           sampleMeta.castString(std::string(metaName.Data())));  
  }
  out.ReplaceAll("/", "");
  return out.Data();
}

merge()

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

static

Definition at line 296 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
 
  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());
  
  const std::string findCmd(Form("find %s -name \"*.%s*\" | tr '\n' ' '", 
                 dir.c_str(), fileName.c_str()));
  std::istringstream input(gSystem->GetFromPipe(findCmd.c_str()).Data()); 
  std::vector<std::string> files((std::istream_iterator<std::string>(input)),
                 std::istream_iterator<std::string>());
  
  std::sort(files.begin(), files.end());
  RCU_ASSERT(std::unique(files.begin(), files.end()) == files.end());
  
  if (not files.size()) {
    std::cerr << "Found no input files for merging! "
          << "Requeueing sample for download..." << std::endl;
    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"); 
    gSystem->Exec(Form("rm -f %s", target.c_str()));
    return Status::PENDING;
  }
 
  for (size_t i = 0; i != files.size(); ++i) {
    gSystem->Exec(Form("rm %s", files[i].c_str()));
  }  
  gSystem->Exec(Form("rmdir %s/*", dir.c_str()));
  gSystem->Exec(Form("rmdir %s", dir.c_str()));
 
  return Status::DONE;
}

nextState()

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

static

Definition at line 109 of file PrunDriver.cxx.

{
  RCU_REQUIRE(state != JobState::FINISHED);
  RCU_REQUIRE(state != JobState::FAILED);
  static const TransitionRule TABLE[] =
    {
      TransitionRule(JobState::INIT,     Status::DONE,    JobState::RUN),
      TransitionRule(JobState::INIT,     Status::PENDING, JobState::INIT),
      TransitionRule(JobState::INIT,     Status::FAIL,    JobState::FAILED),
      TransitionRule(JobState::RUN,      Status::DONE,    JobState::DOWNLOAD),
      TransitionRule(JobState::RUN,      Status::PENDING, JobState::RUN),
      TransitionRule(JobState::RUN,      Status::FAIL,    JobState::FAILED),
      TransitionRule(JobState::DOWNLOAD, Status::DONE,    JobState::MERGE),
      TransitionRule(JobState::DOWNLOAD, Status::PENDING, JobState::DOWNLOAD),
      TransitionRule(JobState::DOWNLOAD, Status::FAIL,    JobState::FAILED),
      TransitionRule(JobState::MERGE,    Status::DONE,    JobState::FINISHED),
      TransitionRule(JobState::MERGE,    Status::PENDING, JobState::MERGE),
      TransitionRule(JobState::MERGE,    Status::FAIL,    JobState::DOWNLOAD)
    };
  static const unsigned int TABLE_SIZE = sizeof(TABLE) / sizeof(TABLE[0]);
  for (unsigned int i = 0; i != TABLE_SIZE; ++i) {
    if (TABLE[i].fromState == state && TABLE[i].status == status) {
      return TABLE[i].toState;
    }
  }
  RCU_ASSERT0("Missing state transition rule");
  throw std::logic_error("PrunDriver.cxx: Missing state transition rule"); 
}

outputFileNames()

std::string outputFileNames

(

const EL::Job &

job

)

Definition at line 437 of file PrunDriver.cxx.

{
  TList outputs;
  for (EL::Job::outputIter out = job.outputBegin(),
     end = job.outputEnd(); out != end; ++out) {
    outputs.Add(out->Clone());
  }      
  std::string out = "hist:hist-output.root";
  TIter itr(&outputs);
  TObject *obj = 0;
  while ((obj = itr())) {
    EL::OutputStream *os = dynamic_cast<EL::OutputStream*>(obj);
    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 481 of file PrunDriver.cxx.

{
  SH::SampleHandler out;
  const std::string outputFile = "*" + outputLabel + ".root*";
  const std::string outDSSuffix = '_' + outputLabel + ".root/"; 
  for (SH::SampleHandler::iterator s = in.begin(); s != in.end(); ++s) {
    auto outSample = std::make_unique<SH::SampleGrid>((*s)->name());
    const std::string outputDS = (*s)->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 377 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sh.size());
 
  WorkList workList;
 
  bool isFirstSample = true;
  for (SH::SampleHandler::iterator s = sh.begin(); s != sh.end(); ++s) {
    if (sampleState(*s) == state) {
      workList.push_back([s, isFirstSample]()->void{ processTask(*s, isFirstSample); });
      // 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 346 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 101 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 458 of file PrunDriver.cxx.

{    
  TFile file(fileName.c_str(), "RECREATE"); 
  TList outputs; 
  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::SampleHandler::iterator s = sh.begin(); s != sh.end(); ++s) {
    const SH::MetaObject& meta = *((*s)->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 179 of file PrunDriver.cxx.

{
  RCU_REQUIRE(sample);
  using namespace EL::msgEventLoop;
 
  ANA_MSG_INFO( "Submitting " << sample->name() << "..." );
 
  static bool loaded = false;
  if (not loaded) {
    // TString path = "$ROOTCOREBIN/python/EventLoopGrid/ELG_prun.py";
    // gSystem->ExpandPathName(path);
    // TPython::LoadMacro(path.Data());
    std::string path = PathResolverFindCalibFile("EventLoopGrid/ELG_prun.py");
    TPython::LoadMacro(path.c_str());
    loaded = true;
  }
 
  TPython::Bind(dynamic_cast<TObject*>(sample), "ELG_SAMPLE"); 
#if ROOT_VERSION_CODE >= ROOT_VERSION(6,33,01)
  std::any result;
  TPython::Exec("_anyresult = ROOT.std.make_any['int'](ELG_prun(ELG_SAMPLE))", &result);
  int ret = std::any_cast<int>(result);
#else
  int ret = TPython::Eval("ELG_prun(ELG_SAMPLE)");
#endif
  TPython::Bind(0, "ELG_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;
}