In This Package:

JobInfoSvc.cc

Go to the documentation of this file.

#include "JobInfoSvc.h"

#include "GaudiKernel/GaudiException.h"
#include "GaudiKernel/MsgStream.h"
#include "Event/JobHeader.h"
#include "Event/JobInfo.h"
#include "RootIOSvc/RootIOAddress.h"
#include "RootIOSvc/IRootIOSvc.h"
#include "GaudiKernel/IConversionSvc.h"
#include "PerJobInfo/PerJobHeader.h"
#include "RootIOSvc/RootIOTypedCnv.h"

JobInfoSvc::JobInfoSvc(const std::string& name,
                             ISvcLocator* svc) :
  Service(name,
          svc),
  m_currentJobInfo(0),
  m_fileIsInitialized(false)
{
  declareProperty("ReadFromFile",m_readFromFile=false,
                  "Load the job info from the current input file?");
  declareProperty("ReadFromDatabase",m_readFromDatabase=true,
                  "Check the database for job info?");
  declareProperty("JobInfoLocation",
                  m_jobInfoLocation = DayaBay::JobHeaderLocation::Default,
                  "Path for job info if reading from file");
  declareProperty("JobId",m_jobIdString,
                  "Used by nuwa.py to initialize current job ID");
  declareProperty("JobConfig",m_jobConfig,
                  "Use to initialize current job configuration");
}

StatusCode JobInfoSvc::queryInterface(const InterfaceID& id,
                                      void** interface) {
  if (IJobInfoSvc::interfaceID().versionMatch(id)) {
    // If good enough match return this object.
    *interface = (IJobInfoSvc*)this;
    addRef();
  } else {
    // Interface is not directly available: try out a base class.
    return Service::queryInterface(id,
                                   interface);
  }

  return StatusCode::SUCCESS;
}

StatusCode JobInfoSvc::initialize() {
  StatusCode status = Service::initialize();
  if (status.isFailure()) {
    throw GaudiException("Could not initialize super class",
                         name(),
                         status);
  }
  m_jobId.set(m_jobIdString);
  return StatusCode::SUCCESS;
}

StatusCode JobInfoSvc::finalize() {
  // Clear job info list
  std::vector<DayaBay::JobInfo*>::iterator it, done = m_jobInfoList.end();
  for (it = m_jobInfoList.begin(); it != done; ++it) {
    // boost::pool somehow owns the job info objects, so don't delete
    //delete *it;
  }
  m_jobInfoList.clear();
  return Service::finalize();
}

const DayaBay::JobInfo* JobInfoSvc::currentJobInfo() {
  // Get the information for the current job
  MsgStream log(msgSvc(), "JobInfoSvc");
  if(!m_currentJobInfo){
    // Construct current job information
    m_currentJobInfo = new DayaBay::JobInfo();
    m_currentJobInfo->setJobId(m_jobId);
    m_currentJobInfo->setParameters(m_jobConfig);
    m_jobInfoList.push_back(m_currentJobInfo);
  }
  return m_currentJobInfo;
}

const DayaBay::JobInfo* JobInfoSvc::jobInfo(const DayaBay::JobId& jobId) {
  // Return the job info given the job ID

  MsgStream log(msgSvc(), "JobInfoSvc");
  StatusCode sc = StatusCode::SUCCESS;
  // Check if file info should be loaded from file
  if( m_readFromFile && !m_fileIsInitialized){
    sc = this->readDataFromFile();
    if(!sc.isSuccess()) return 0;
  }

  // Check current cache for this job ID
  DayaBay::JobInfo* jobInfo = 0;
  std::vector<DayaBay::JobInfo*>::iterator jobIter, 
    jobDone = m_jobInfoList.end();
  for(jobIter = m_jobInfoList.begin(); jobIter != jobDone; ++jobIter) {
    DayaBay::JobInfo* checkJob = *jobIter;
    log << MSG::DEBUG << "Checking job info with ID " 
        << checkJob->jobId().uuid() << endreq;
    if(checkJob->jobId()==jobId){
      // Fond the job
      jobInfo = checkJob;
      break;
    }
  }

  // Not in cache?  Check database
  if(!jobInfo && m_readFromDatabase ){
    sc = this->readDataFromDatabase(jobId);
    if(!sc.isSuccess()) return 0;
  }

  return jobInfo;
}

StatusCode JobInfoSvc::setJobInfo(const DayaBay::JobInfo& jobInfo){
  // Add or update the job info for the given job
  bool updateJob = false;
  std::vector<DayaBay::JobInfo*>::iterator jobIter, 
    jobDone = m_jobInfoList.end();
  for(jobIter = m_jobInfoList.begin(); jobIter != jobDone; ++jobIter) {
    if( (*jobIter)->jobId() == jobInfo.jobId() ){
      // Found job in current list; update
      *(*jobIter) = jobInfo;
      updateJob = true;
      break;
    }
  }
  // Catch new job
  if( !updateJob ) m_jobInfoList.push_back(new DayaBay::JobInfo(jobInfo));
  return StatusCode::SUCCESS;
}

const std::vector<DayaBay::JobInfo*>& JobInfoSvc::cachedJobInfo(){
  // Return the current internal list of job info.
  // Use this for Database / File storage of job info

  // Ensure current job info is initialized
  if(!m_currentJobInfo){
    this->currentJobInfo();
  }
  // Ensure input job info is initialized, if needed
  if(m_readFromFile && !m_fileIsInitialized){
    StatusCode sc = this->readDataFromFile();
    sc.ignore();
  }
  return m_jobInfoList;
}

StatusCode JobInfoSvc::readDataFromDatabase(const DayaBay::JobId& /*jobId*/){
  // Read the job info from the Database
  // FIXME: ADD database code here
  MsgStream log(msgSvc(), "JobInfoSvc");
  log << MSG::WARNING << "readDataFromDatabase: Loading data from database is not yet implemented." << endreq;
  return StatusCode::SUCCESS;
}

StatusCode JobInfoSvc::readDataFromFile(){
  // Read job info from current input file

  // If file already read, don't read again
  if( m_fileIsInitialized ) return StatusCode::SUCCESS;

  MsgStream log(msgSvc(), "JobInfoSvc");
  IService* isvc = 0; 
  StatusCode sc = this->service("RootIOCnvSvc",isvc,true);
  if(sc.isFailure()){
    log << MSG::ERROR << "Failed to get RootIOCnvSvc as IService" << endreq;
    return sc;
  }
  IRootIOSvc* m_rioSvc = 0;
  sc = isvc->queryInterface(IRootIOSvc::interfaceID(), (void**)&m_rioSvc);
  if (sc.isFailure()) {
    log << MSG::ERROR << "Conversion service RootIOCnvSvc"
        << " does not implement IRootIOCnvSvc" << endreq;
    return sc;
  }
  IConversionSvc* convSvc = 0;
  sc = isvc->queryInterface(IConversionSvc::interfaceID(), (void**)&convSvc);
  if (sc.isFailure()) {
    log << MSG::ERROR << "Conversion service RootIOCnvSvc"
        << " does not implement IConversionSvc" << endreq;
    return sc;
  }
  IRootIOSvc::InputStreamMap& ism = m_rioSvc->inputStreams();
  IRootIOSvc::InputStreamMap::iterator it, done = ism.end();
  for (it = ism.begin(); it != done; ++it) {
    log << MSG::DEBUG << "checking input stream: " << it->first << endreq;
    if( it->first == m_jobInfoLocation ){
      // Get Persistent JobHeader from file
      RootInputStream* ris = it->second;
      ris->setEntry(0);
      log << MSG::DEBUG << "Return status of read: " << ris->read() << "\t" 
          << ris->obj() << endreq;
      PerJobHeader* perJobHeader = (PerJobHeader*)(ris->obj());
      if(!perJobHeader){
        log << MSG::ERROR << "Failed to get persistent JobHeader" << endreq;
        return StatusCode::FAILURE;
      }
      log << MSG::DEBUG << "Got persistent job info for "
          << perJobHeader->jobInfoList.size() << " jobs" << endreq;
      // Convert Persistent JobHeader to Transient
      //DayaBay::JobHeader jobHeader;
      DayaBay::JobHeader* jobHeader = new DayaBay::JobHeader();
      IConverter* converter = convSvc->converter(jobHeader->classID());
      if (!converter) {
        log << MSG::ERROR << "Failed to get JobHeader converter" << endreq;
        return StatusCode::FAILURE;
      }
      RootIOTypedCnv<PerJobHeader, DayaBay::JobHeader>* jobHdrCnv = 0;
      jobHdrCnv = dynamic_cast< RootIOTypedCnv<PerJobHeader, DayaBay::JobHeader>* >(converter);
      if (!jobHdrCnv) {
        log << MSG::ERROR << "Failed to cast JobHeader converter" << endreq;
        return StatusCode::FAILURE;
      }
      sc = jobHdrCnv->PerToTran(*perJobHeader, *jobHeader);
      if (sc.isFailure()) {
        log << MSG::ERROR << "Failed to translate persistent JobHeader" 
            << endreq;
        return sc;
      }
      const DayaBay::JobHeader::JobInfoContainer& jobInfoList 
        = jobHeader->jobInfoList();
      int nJobs = jobInfoList.size();
      log << MSG::DEBUG << "Got job info for " << nJobs << " jobs" << endreq;
      for(int jobIdx=0; jobIdx < nJobs; jobIdx++){
        // Add each job info to the current list
        m_jobInfoList.push_back(new DayaBay::JobInfo(*(jobInfoList[jobIdx])));
      }
    }
  }
  m_fileIsInitialized = true;
  return StatusCode::SUCCESS;
}

---