EL::PrunDriver Class Reference

a Driver to submit jobs via prun More...

#include <PrunDriver.h>

Inheritance diagram for EL::PrunDriver:

Collaboration diagram for EL::PrunDriver:

Public Member Functions
	PrunDriver ()
void	testInvariant () const
SH::MetaObject *	options ()
	the list of options to jobs with this driver
const SH::MetaObject *	options () const
std::string	submit (const Job &job, const std::string &location) const
	submit the given job with the given output location and wait for it to finish
std::string	submitOnly (const Job &job, const std::string &location) const
	submit the given job with the given output location and return immediately

Static Public Member Functions
static void	status (const std::string &location)
static void	setState (const std::string &location, const std::string &task, const std::string &state)
static void	resubmit (const std::string &location, const std::string &option)
	resubmit all failed sub-jobs for the job in the given location
static bool	retrieve (const std::string &location)
	retrieve all the output for the job in the given location
static bool	wait (const std::string &location, unsigned time=60)
	retrieve all the output for the job in the given location and wait until it is finished completely.
static void	updateLocation (const std::string &location)
	update the internal location of files, after moving the submission directory
static void	mergedOutputSave (Detail::ManagerData &data)
	create and save a sample handler assuming we created all the merged files at the requested locations
static void	diskOutputSave (Detail::ManagerData &data)
	make the output sample handler for the given job or stream from the information stored in the histogram files.

Protected Member Functions
virtual::StatusCode	doManagerStep (Detail::ManagerData &data) const override

Static Protected Attributes
static bool	abortRetrieve
	this flag is set to true when the wait() function is running and a SIGINT is caught, meaning that control should be returned to the user as soon as possible.

Private Member Functions
::StatusCode	doRetrieve (Detail::ManagerData &data) const
	ClassDef (EL::PrunDriver, 1)

Private Attributes
SH::MetaObject	m_options
	members directly corresponding to accessors

Detailed Description

a Driver to submit jobs via prun

Definition at line 19 of file PrunDriver.h.

Constructor & Destructor Documentation

PrunDriver()

EL::PrunDriver::PrunDriver

(

)

Definition at line 591 of file PrunDriver.cxx.

{
  RCU_NEW_INVARIANT(this);
}

Member Function Documentation

ClassDef()

EL::PrunDriver::ClassDef ( EL::PrunDriver , 1  )

private

diskOutputSave()

void EL::Driver::diskOutputSave ( Detail::ManagerData & data )

staticinherited

make the output sample handler for the given job or stream from the information stored in the histogram files.

This is optional, but it is convenient for drivers that use (conventional) writers

Guarantee

basic

Failures

out of memory II
i/o errors

doManagerStep()

StatusCode EL::PrunDriver::doManagerStep ( Detail::ManagerData & data ) const

overrideprotected

Definition at line 596 of file PrunDriver.cxx.

{
  using namespace msgEventLoop;
  ANA_CHECK (Driver::doManagerStep (data));
  switch (data.step)
  {
  case Detail::ManagerStep::submitJob:
    {
      const std::string jobELGDir = data.submitDir + "/elg";
      const std::string runShFile = jobELGDir + "/runjob.sh";
      const std::string mergeShFile = jobELGDir + "/elg_merge";
      const std::string runShOrig = PathResolverFindCalibFile("EventLoopGrid/runjob.sh");
      const std::string mergeShOrig = PathResolverFindCalibFile("EventLoopGrid/elg_merge");
 
      const std::string jobDefFile = jobELGDir + "/jobdef.root";
 
      namespace fs = std::filesystem;
      std::error_code ec;
      fs::create_directories(jobELGDir, ec);
      if (ec) {
        ANA_MSG_ERROR("could not create directory " << jobELGDir << ": " << ec.message());
        return StatusCode::FAILURE;
      }
      // Copy the grid scripts into the submission directory and make them
      // executable, aborting submission if either step fails.
      const auto copyExecutable =
        [&] (const std::string& from, const std::string& to) -> StatusCode {
          std::error_code ec2;
          fs::copy_file(from, to, fs::copy_options::overwrite_existing, ec2);
          if (ec2) {
            ANA_MSG_ERROR("could not copy " << from << " to " << to << ": " << ec2.message());
            return StatusCode::FAILURE;
          }
          fs::permissions(to, fs::perms::owner_exec | fs::perms::group_exec |
                          fs::perms::others_exec, fs::perm_options::add, ec2);
          if (ec2) {
            ANA_MSG_ERROR("could not make " << to << " executable: " << ec2.message());
            return StatusCode::FAILURE;
          }
          return StatusCode::SUCCESS;
        };
      ANA_CHECK(copyExecutable(runShOrig, runShFile));
      ANA_CHECK(copyExecutable(mergeShOrig, mergeShFile));
 
      // create symbolic links for additionnal files/directories if any to ship to the grid 
      std::string listToShipToGrid = data.options.castString(EL::Job::optGridPrunShipAdditionalFilesOrDirs, ""); 
      // parse the list of comma separated files and/or directories to ship to the grid 
      if (listToShipToGrid.size()){
        ANA_MSG_INFO (
          "Creating symbolic links for additional files or directories to be sent to grid.\n"
          "For root or heavy files you should also add their name (not the full path) to EL::Job::optUserFiles.\n"
          "Otherwise prun ignores those files."
        );
 
        std::vector<std::string> vect_filesOrDirToShip;
        for (auto&& part : std::views::split(listToShipToGrid, ',')) vect_filesOrDirToShip.emplace_back(part.begin(), part.end());
        // Create symbolic links of files or directories to the submission directory
        for (const std::string & fileOrDirToShip: vect_filesOrDirToShip){
          ANA_MSG_INFO (("Creating symbolic link for: " +fileOrDirToShip).c_str());
          const fs::path linkPath =
            fs::path(jobELGDir) / fs::path(fileOrDirToShip).filename();
          // emulate `ln -sf`: replace any pre-existing link/file
          fs::remove(linkPath, ec);
          fs::create_symlink(fileOrDirToShip, linkPath, ec);
          if (ec) {
            ANA_MSG_ERROR("could not create symbolic link " << linkPath.string()
                          << " -> " << fileOrDirToShip << ": " << ec.message());
            return StatusCode::FAILURE;
          }
        }
        ANA_MSG_INFO ("Finished creation of symbolic links");
      }
 
      const SH::SampleHandler& sh = data.job->sampleHandler();
 
      for (SH::Sample* const sample : sh) {
        SH::MetaObject& meta = *sample->meta();
        meta.fetchDefaults(data.options);
        meta.fetchDefaults(defaultOpts());
        meta.setString("nc_outputs", outputFileNames(*data.job));
        std::string outputSampleName = meta.castString("nc_outputSampleName");
        if (outputSampleName.empty()) {
          outputSampleName = "user.%nickname%.%in:name%";
        }
        meta.setString("nc_outDS", formatOutputName(meta, outputSampleName));
        meta.setString("nc_inDS", meta.castString("nc_grid", sample->name()));
        meta.setString("nc_writeInputToTxt", "IN:input.txt");
        meta.setString("nc_match", meta.castString("nc_grid_filter"));
        const std::string execstr = "runjob.sh " + sample->name();
        meta.setString("nc_exec", execstr);
        meta.setString("nc_framework", "EventLoopGrid");
      }
 
      saveJobDef(jobDefFile, *data.job, sh);
  
      for (EL::Job::outputIter out = data.job->outputBegin();
           out != data.job->outputEnd(); ++out) {
        SH::SampleHandler shOut = outputSH(sh, out->label());
        shOut.save(data.submitDir + "/output-" + out->label());
      }
      SH::SampleHandler shHist = outputSH(sh, "hist-output");
      shHist.save(data.submitDir + "/output-hist");
 
      TmpCd keepDir(jobELGDir);
 
      processAllInState(sh, JobState::INIT, 0); 
 
      sh.save(data.submitDir + "/input");
      data.submitted = true;
    }
    break;
 
  case Detail::ManagerStep::doRetrieve:
    {
      ANA_CHECK (doRetrieve (data));
    }
    break;
 
  default:
    (void) true; // safe to do nothing
  }
  return ::StatusCode::SUCCESS;
}

doRetrieve()

StatusCode EL::PrunDriver::doRetrieve ( Detail::ManagerData & data ) const

private

Definition at line 721 of file PrunDriver.cxx.

{
  RCU_READ_INVARIANT(this);
  RCU_REQUIRE(not data.submitDir.empty());  
 
  TmpCd tmpDir(data.submitDir);
 
  SH::SampleHandler sh;
  sh.load("input");
  RCU_ASSERT(sh.size());
 
  const size_t nRunThreads = options()->castDouble("nc_run_threads", 0); 
  const size_t nDlThreads = options()->castDouble("nc_download_threads", 0); 
  processAllInState(sh, JobState::INIT, 0); 
  processAllInState(sh, JobState::RUN, nRunThreads);
  processAllInState(sh, JobState::DOWNLOAD, nDlThreads); 
  processAllInState(sh, JobState::MERGE, 0); 
 
  sh.save("input");
 
  std::cout << std::endl;
 
  bool allDone = true;
  for (SH::Sample* const sample : sh) {
    JobState::Enum state = sampleState(sample);
    std::string details = sample->meta()->castString("nc_ELG_state_details", "", SH::MetaObject::CAST_NOCAST_DEFAULT);
    if (not details.empty()) { details = '(' + details + ')'; }
 
    std::cout << sample->name() << "\t";
    switch (state) {
    case JobState::INIT:
    case JobState::RUN:
    case JobState::DOWNLOAD:
    case JobState::MERGE:
      std::cout << JobState::name[state] << "\t";
      break;
    case JobState::FINISHED:
      std::cout << "\033[1;32m" << JobState::name[state] << "\033[0m\t";
      break;
    case JobState::FAILED:
      std::cout << "\033[1;31m" << JobState::name[state] << "\033[0m\t";
      break;
    }
    std::cout << details << std::endl;
    
    allDone &= (state == JobState::FINISHED || state == JobState::FAILED);
  }
 
  std::cout << std::endl;
  
  data.retrieved = true;
  data.completed = allDone;
  return ::StatusCode::SUCCESS;
}

mergedOutputSave()

void EL::Driver::mergedOutputSave ( Detail::ManagerData & data )

staticinherited

create and save a sample handler assuming we created all the merged files at the requested locations

This is optional, but it is convenient for drivers that want to keep their outputs locally.

Guarantee

basic

Failures

out of memory II
i/o errors

options() [1/2]

SH::MetaObject * EL::Driver::options ( )

inherited

the list of options to jobs with this driver

Guarantee

no-fail

Postcondition

result != 0

options() [2/2]

const SH::MetaObject * EL::Driver::options ( ) const

inherited

resubmit()

void EL::Driver::resubmit ( const std::string & location, const std::string & option )

staticinherited

resubmit all failed sub-jobs for the job in the given location

\parm option driver-specific option string selecting which jobs to resubmit (and how)

Guarantee

basic, may partially resubmit

Failures

out of memory III
job resubmission errors
job can't be read
job was made with different driver

retrieve()

bool EL::Driver::retrieve ( const std::string & location )

staticinherited

retrieve all the output for the job in the given location

While job failures will cause this method to fail you can typically retry it multiple times if you can use partial results.

Returns

whether the job completed successfully

Guarantee

basic, may partially retrieve

Failures

out of memory III
job failures
job can't be read
job was made with different driver

setState()

void EL::PrunDriver::setState ( const std::string & location, const std::string & task, const std::string & state )

static

Definition at line 794 of file PrunDriver.cxx.

{
  RCU_REQUIRE(not location.empty());  
  RCU_REQUIRE(not task.empty());  
  RCU_REQUIRE(not state.empty());  
  TmpCd tmpDir(location);
  SH::SampleHandler sh;
  sh.load("input");
  RCU_ASSERT(sh.size());
  if (not sh.get(task)) {
    std::cout << "Unknown task: " << task << std::endl;
    std::cout << "Choose one of: " << std::endl;
    sh.print();
    return;
  }
  JobState::parse(state);
  sh.get(task)->meta()->setString("nc_ELG_state", state);
  sh.save("input");
}

status()

void EL::PrunDriver::status ( const std::string & location )

static

Definition at line 776 of file PrunDriver.cxx.

{
  RCU_REQUIRE(not location.empty());  
  TmpCd tmpDir(location);
  SH::SampleHandler sh;
  sh.load("input");
  RCU_ASSERT(sh.size());
  processAllInState(sh, JobState::RUN, 0); 
  sh.save("input");
  for (SH::Sample* const sample : sh) {
    JobState::Enum state = sampleState(sample);
    std::string details = sample->meta()->castString("nc_ELG_state_details", "", SH::MetaObject::CAST_NOCAST_DEFAULT);
    if (not details.empty()) { details = '(' + details + ')'; }
    std::cout << sample->name() << "\t" << JobState::name[state]
          << "\t" << details << std::endl;
  }
}

submit()

std::string EL::Driver::submit ( const Job & job, const std::string & location ) const

inherited

submit the given job with the given output location and wait for it to finish

This is mostly for small jobs and backward compatibility. For longer jobs use submitOnly instead.

Returns

The actual location of the submit directory, if the job was configured to generate a unique directory.

Guarantee

basic, may partially submit

Failures

out of memory II

Failures

can't create directory at location
submission errors

submitOnly()

std::string EL::Driver::submitOnly ( const Job & job, const std::string & location ) const

inherited

submit the given job with the given output location and return immediately

This method allows you to submit jobs to your local batch system, log out and at a later point log back in again.

Returns

The actual location of the submit directory, if the job was configured to generate a unique directory.

Guarantee

basic, may partially submit

Failures

out of memory II
can't create directory at location
submission errors \warn not all drivers support this. some will do all their work in the submit function. \warn you normally need to call wait() or retrieve() before you can use the output.

testInvariant()

void EL::PrunDriver::testInvariant

(

)

const

Definition at line 588 of file PrunDriver.cxx.

{}

updateLocation()

void EL::Driver::updateLocation ( const std::string & location )

staticinherited

update the internal location of files, after moving the submission directory

Guarantee

basic, may update partially

Failures

out of memory II \warn only move the submission directory after all your jobs are finished, or the results will be unpredictable

wait()

bool EL::Driver::wait ( const std::string & location, unsigned time = 60 )

staticinherited

retrieve all the output for the job in the given location and wait until it is finished completely.

poll the output every time seconds.

While job failures will cause this method to fail you can typically retry it multiple times if you can use partial results.

Typically sleeping for 60 seconds is an appropriate interval, but if it doesn't work for you, you can change it here.

Guarantee

basic, may partially retrieve

Failures

out of memory III
job failures
job can't be read
job was made with different driver

Member Data Documentation

abortRetrieve

bool EL::Driver::abortRetrieve

staticprotectedinherited

this flag is set to true when the wait() function is running and a SIGINT is caught, meaning that control should be returned to the user as soon as possible.

drivers can use it to abort long running operations in doRetrieve before completion

Definition at line 212 of file Driver.h.

m_options

SH::MetaObject EL::Driver::m_options

privateinherited

members directly corresponding to accessors

Definition at line 233 of file Driver.h.

---

The documentation for this class was generated from the following files:

• PrunDriver.h
• PrunDriver.cxx