evf::EvFDaqDirector Class Reference

#include <EvFDaqDirector.h>

Public Types
enum	FileStatus {   noFile, sameFile, newFile, newLumi,   runEnded, runAbort }

Public Member Functions
std::string &	baseRunDir ()
std::string &	buBaseRunDir ()
std::string &	buBaseRunOpenDir ()
void	createRunOpendirMaybe ()
	EvFDaqDirector (const edm::ParameterSet &pset, edm::ActivityRegistry &reg)
std::string	findCurrentRunDir ()
std::string	getEoLSFilePathOnBU (const unsigned int ls) const
std::string	getEoLSFilePathOnFU (const unsigned int ls) const
std::string	getEoRFilePath () const
std::string	getEoRFilePathOnFU () const
std::string	getInitFilePath (std::string const &stream) const
std::string	getInputJsonFilePath (const unsigned int ls, const unsigned int index) const
std::string	getJumpFilePath () const
unsigned int	getJumpIndex () const
unsigned int	getJumpLS () const
std::string	getMergedDatChecksumFilePath (const unsigned int ls, std::string const &stream) const
std::string	getMergedDatFilePath (const unsigned int ls, std::string const &stream) const
std::string	getMergedProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getMergedRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenDatFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenInputJsonFilePath (const unsigned int ls, const unsigned int index) const
std::string	getOpenOutputJsonFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenRawFilePath (const unsigned int ls, const unsigned int index) const
std::string	getOpenRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOutputJsonFilePath (const unsigned int ls, std::string const &stream) const
std::string	getProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getRawFilePath (const unsigned int ls, const unsigned int index) const
std::string	getRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
unsigned int	getRunNumber () const
std::string	getRunOpenDirPath () const
std::vector< int > *	getStreamFileTracker ()
bool	getTestModeNoBuilderUnit ()
bool	isSingleStreamThread ()
void	lockFULocal ()
void	lockFULocal2 ()
void	lockInitLock ()
FILE *	maybeCreateAndLockFileHeadForStream (unsigned int ls, std::string &stream)
bool	outputAdler32Recheck () const
void	postEndRun (edm::GlobalContext const &globalContext)
void	preallocate (edm::service::SystemBounds const &bounds)
void	preBeginRun (edm::GlobalContext const &globalContext)
void	preGlobalEndLumi (edm::GlobalContext const &globalContext)
void	preSourceEvent (edm::StreamID const &streamID)
int	readLastLSEntry (std::string const &file)
void	removeFile (unsigned int ls, unsigned int index)
void	removeFile (std::string)
void	setDeleteTracking (std::mutex *fileDeleteLock, std::list< std::pair< int, InputFile * >> *filesToDelete)
void	setFMS (evf::FastMonitoringService *fms)
void	tryInitializeFuLockFile ()
void	unlockAndCloseMergeStream ()
void	unlockFULocal ()
void	unlockFULocal2 ()
void	unlockInitLock ()
void	updateFileIndex (int const &fileIndex)
FileStatus	updateFuLock (unsigned int &ls, std::string &nextFile, uint32_t &fsize)
	~EvFDaqDirector ()

Private Member Functions
bool	bumpFile (unsigned int &ls, unsigned int &index, std::string &nextFile, uint32_t &fsize, int maxLS)
std::string	eolsFileName (const unsigned int ls) const
std::string	eorFileName () const
int	getNFilesFromEoLS (std::string BUEoLSFile)
std::string	initFileName (std::string const &stream) const
std::string	inputFileNameStem (const unsigned int ls, const unsigned int index) const
std::string	mergedFileNameStem (const unsigned int ls, std::string const &stream) const
void	openFULockfileStream (std::string &fuLockFilePath, bool create)
std::string	outputFileNameStem (const unsigned int ls, std::string const &stream) const

Private Attributes
std::string	base_dir_
std::string	bu_base_dir_
struct flock	bu_r_flk
struct flock	bu_r_fulk
FILE *	bu_r_lock_stream
int	bu_readlock_fd_
std::string	bu_run_dir_
std::string	bu_run_open_dir_
FILE *	bu_t_monitor_stream
struct flock	bu_w_flk
struct flock	bu_w_fulk
FILE *	bu_w_lock_stream
FILE *	bu_w_monitor_stream
int	bu_writelock_fd_
int	currentFileIndex_ = -1
int	data_readwrite_fd_
struct flock	data_rw_flk
struct flock	data_rw_fulk
FILE *	data_rw_stream
bool	directorBu_
DirManager	dirManager_
unsigned int	eolsNFilesIndex_ = 1
std::mutex *	fileDeleteLockPtr_ = nullptr
std::list< std::pair< int, InputFile * > > *	filesToDeletePtr_ = nullptr
evf::FastMonitoringService *	fms_ = nullptr
int	fu_readwritelock_fd_
struct flock	fu_rw_flk
struct flock	fu_rw_fulk
FILE *	fu_rw_lock_stream
int	fulocal_rwlock_fd2_
int	fulocal_rwlock_fd_
std::string	hostname_
pthread_mutex_t	init_lock_ = PTHREAD_MUTEX_INITIALIZER
unsigned int	jumpIndex_
unsigned int	jumpLS_
unsigned int	nStreams_ =0
unsigned int	nThreads_ =0
bool	outputAdler32Recheck_
unsigned long	previousFileSize_
bool	readEolsDefinition_ = true
unsigned int	run_
std::string	run_dir_
std::string	run_string_
std::string	stopFilePath_
std::vector< int >	streamFileTracker_
bool	testModeNoBuilderUnit_

Detailed Description

Definition at line 38 of file EvFDaqDirector.h.

Member Enumeration Documentation

enum evf::EvFDaqDirector::FileStatus

Enumerator
noFile
sameFile
newFile
newLumi
runEnded
runAbort

Definition at line 42 of file EvFDaqDirector.h.

{ noFile, sameFile, newFile, newLumi, runEnded, runAbort };

Constructor & Destructor Documentation

evf::EvFDaqDirector::EvFDaqDirector ( const edm::ParameterSet &  pset, edm::ActivityRegistry &  reg  )

explicit

Definition at line 38 of file EvFDaqDirector.cc.

References base_dir_, bu_base_dir_, bu_run_dir_, bu_run_open_dir_, bu_w_lock_stream, bu_writelock_fd_, createRunOpendirMaybe(), directorBu_, edm::hlt::Exception, fu_readwritelock_fd_, fu_rw_lock_stream, fulocal_rwlock_fd2_, fulocal_rwlock_fd_, getRunOpenDirPath(), hostname_, init_lock_, NULL, openFULockfileStream(), postEndRun(), preallocate(), preBeginRun(), preGlobalEndLumi(), preSourceEvent(), run_, run_dir_, run_string_, contentValuesCheck::ss, stopFilePath_, AlCaHLTBitMon_QueryRunRegistry::string, tryInitializeFuLockFile(), edm::ActivityRegistry::watchPostGlobalEndRun(), edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreGlobalBeginRun(), edm::ActivityRegistry::watchPreGlobalEndLumi(), and edm::ActivityRegistry::watchPreSourceEvent().

                                           :
    testModeNoBuilderUnit_(
               pset.getUntrackedParameter<bool> ("testModeNoBuilderUnit",
                                 false)
               ),
    base_dir_(
          pset.getUntrackedParameter<std::string> ("baseDir", "/data")
          ),
    bu_base_dir_(
         pset.getUntrackedParameter<std::string> ("buBaseDir", "/data")
         ),
    directorBu_(
        pset.getUntrackedParameter<bool> ("directorIsBu", false)
        ),
    run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
    outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck",false)),
    hostname_(""),
    bu_readlock_fd_(-1),
    bu_writelock_fd_(-1),
    fu_readwritelock_fd_(-1),
    data_readwrite_fd_(-1),
    fulocal_rwlock_fd_(-1),
    fulocal_rwlock_fd2_(-1),

    bu_w_lock_stream(0),
    bu_r_lock_stream(0),
    fu_rw_lock_stream(0),
    //bu_w_monitor_stream(0),
    //bu_t_monitor_stream(0),
    data_rw_stream(0),

    dirManager_(base_dir_),

    previousFileSize_(0),
    jumpLS_(0),
    jumpIndex_(0),

    bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
    bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
    bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
    bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
    fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
    fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
    data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
    data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
    //fulocal_rw_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
    //fulocal_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
    //fulocal_rw_flk2( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
    //fulocal_rw_fulk2( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
  {

    reg.watchPreallocate(this, &EvFDaqDirector::preallocate);
    reg.watchPreGlobalBeginRun(this, &EvFDaqDirector::preBeginRun);
    reg.watchPostGlobalEndRun(this, &EvFDaqDirector::postEndRun);
    reg.watchPreSourceEvent(this, &EvFDaqDirector::preSourceEvent);
    reg.watchPreGlobalEndLumi(this,&EvFDaqDirector::preGlobalEndLumi);

    std::stringstream ss;
    ss << "run" << std::setfill('0') << std::setw(6) << run_;
    run_string_ = ss.str();
    run_dir_ = base_dir_+"/"+run_string_;

    //save hostname for later
    char hostname[33];
    gethostname(hostname,32);
    hostname_ = hostname;
    // check if base dir exists or create it accordingly
    int retval = mkdir(base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error checking for base dir -: "
                          << base_dir_ << " mkdir error:" << strerror(errno);
    }

    //create run dir in base dir
    umask(0);
    retval = mkdir(run_dir_.c_str(),
               S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error creating run dir -: "
                      << run_dir_ << " mkdir error:" << strerror(errno);
    }

    //create fu-local.lock in run open dir
    if (!directorBu_) {

      createRunOpendirMaybe();
      std::string fulocal_lock_ = getRunOpenDirPath() +"/fu-local.lock";
      fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
      if (fulocal_rwlock_fd_==-1)
        throw cms::Exception("DaqDirector") << " Error creating/opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
      chmod(fulocal_lock_.c_str(),0777);
      fsync(fulocal_rwlock_fd_);
      //open second fd for another input source thread
      fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
      if (fulocal_rwlock_fd2_==-1)
        throw cms::Exception("DaqDirector") << " Error opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
    }

    //bu_run_dir: for FU, for which the base dir is local and the BU is remote, it is expected to be there
    //for BU, it is created at this point
    if (directorBu_)
      {
    bu_run_dir_ = base_dir_ + "/" + run_string_;
    std::string bulockfile = bu_run_dir_ + "/bu.lock";
    std::string fulockfile = bu_run_dir_ + "/fu.lock";

    //make or find bu run dir
    retval = mkdir(bu_run_dir_.c_str(),
               S_IRWXU | S_IRWXG | S_IRWXO);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector")
        << " Error creating bu run dir -: " << bu_run_dir_
        << " mkdir error:" << strerror(errno) << "\n";
    }
    bu_run_open_dir_ = bu_run_dir_ + "/open";
    retval = mkdir(bu_run_open_dir_.c_str(),
               S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error creating bu run open dir -: "
                          << bu_run_open_dir_ << " mkdir error:" << strerror(errno)
                          << "\n";
    }

    // the BU director does not need to know about the fu lock
    bu_writelock_fd_ = open(bulockfile.c_str(),
                O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
    if (bu_writelock_fd_ == -1)
      edm::LogWarning("EvFDaqDirector") << "problem with creating filedesc for buwritelock -: "
                        << strerror(errno);
    else
      edm::LogInfo("EvFDaqDirector") << "creating filedesc for buwritelock -: "
                     << bu_writelock_fd_;
    bu_w_lock_stream = fdopen(bu_writelock_fd_, "w");
    if (bu_w_lock_stream == 0)
      edm::LogWarning("EvFDaqDirector")<< "Error creating write lock stream -: " << strerror(errno);

    // BU INITIALIZES LOCK FILE
    // FU LOCK FILE OPEN
    openFULockfileStream(fulockfile, true);
    tryInitializeFuLockFile();
    fflush(fu_rw_lock_stream);
    close(fu_readwritelock_fd_);
      }
    else
      {
    // for FU, check if bu base dir exists

    retval = mkdir(bu_base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
    if (retval != 0 && errno != EEXIST) {
      throw cms::Exception("DaqDirector") << " Error checking for bu base dir -: "
                          << bu_base_dir_ << " mkdir error:" << strerror(errno) << "\n";
    }

    bu_run_dir_ = bu_base_dir_ + "/" + run_string_;
    std::string fulockfile = bu_run_dir_ + "/fu.lock";
    openFULockfileStream(fulockfile, false);
      }

    pthread_mutex_init(&init_lock_,NULL);

    stopFilePath_ = run_dir_+"/CMSSW_STOP";
  }

evf::EvFDaqDirector::~EvFDaqDirector

(

)

Definition at line 202 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_, fulocal_rwlock_fd_, unlockFULocal(), and unlockFULocal2().

  {
    if (fulocal_rwlock_fd_!=-1) {
      unlockFULocal();
      close(fulocal_rwlock_fd_);
    }

    if (fulocal_rwlock_fd2_!=-1) {
      unlockFULocal2();
      close(fulocal_rwlock_fd2_);
    }

  }

Member Function Documentation

std::string& evf::EvFDaqDirector::baseRunDir ( )

inline

Definition at line 52 of file EvFDaqDirector.h.

References run_dir_.

{return run_dir_;}

std::string& evf::EvFDaqDirector::buBaseRunDir ( )

inline

Definition at line 53 of file EvFDaqDirector.h.

References bu_run_dir_.

{return bu_run_dir_;}

std::string& evf::EvFDaqDirector::buBaseRunOpenDir ( )

inline

Definition at line 54 of file EvFDaqDirector.h.

References bu_run_open_dir_.

{return bu_run_open_dir_;}

bool evf::EvFDaqDirector::bumpFile ( unsigned int &  ls, unsigned int &  index, std::string &  nextFile, uint32_t &  fsize, int  maxLS  )

private

Definition at line 585 of file EvFDaqDirector.cc.

References evf::FastMonitoringService::accumulateFileSize(), fms_, getEoLSFilePathOnBU(), getInputJsonFilePath(), getNFilesFromEoLS(), cmsHarvester::index, python.rootplot.utilities::ls(), previousFileSize_, contentValuesCheck::ss, AlCaHLTBitMon_QueryRunRegistry::string, and testModeNoBuilderUnit_.

Referenced by updateFuLock().

                                                                                                                      {

    if (previousFileSize_ != 0) {
      if (!fms_) {
        try {
          fms_ = (FastMonitoringService *) (edm::Service<evf::MicroStateService>().operator->());
        } catch (...) {
            edm::LogError("EvFDaqDirector") <<" FastMonitoringService not found";
        }
      }
      if (fms_) fms_->accumulateFileSize(ls, previousFileSize_);
      previousFileSize_ = 0;
    }

    struct stat buf;
    std::stringstream ss;
    unsigned int nextIndex = index;
    nextIndex++;

    // 1. Check suggested file
    nextFile = getInputJsonFilePath(ls,index);
    if (stat(nextFile.c_str(), &buf) == 0) {
     
      previousFileSize_ = buf.st_size;
      fsize = buf.st_size;
      return true;
    }
    // 2. No file -> lumi ended? (and how many?)
    else {
      std::string BUEoLSFile = getEoLSFilePathOnBU(ls);
      bool eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
      unsigned int startingLumi = ls;
      while (eolFound) {

        // recheck that no raw file appeared in the meantime
        if (stat(nextFile.c_str(), &buf) == 0) {
          previousFileSize_ = buf.st_size;
          fsize = buf.st_size;
          return true;
        }

        int indexFilesInLS = getNFilesFromEoLS(BUEoLSFile);
        if (indexFilesInLS < 0)
          //parsing failed
          return false;
        else {
          //check index
          if ((int)index<indexFilesInLS) {
            //we have 2 files, and check for 1 failed... retry (2 will never be here)
            edm::LogError("EvFDaqDirector") << "Potential miss of index file in LS -: " << ls << ". Missing "
                         << nextFile << " because " << indexFilesInLS-1 << " is the highest index expected. Will not update fu.lock file";
            return false;
          }
        }
    // this lumi ended, check for files
    ++ls;
        index = 0;

        //reached limit
        if (maxLS>=0 && ls > (unsigned int)maxLS) return false; 

    nextFile = getInputJsonFilePath(ls,0);
    if (stat(nextFile.c_str(), &buf) == 0) {
      // a new file was found at new lumisection, index 0
      previousFileSize_ = buf.st_size;
          fsize = buf.st_size;

      if (testModeNoBuilderUnit_) {
        // rename ended lumi to + 2
            std::string sourceEol = getEoLSFilePathOnBU(startingLumi);

        std::string destEol = getEoLSFilePathOnBU(startingLumi+2);

        std::string cpCmd = "cp " + sourceEol + " " + destEol;
        edm::LogInfo("EvFDaqDirector") << " testmode: Running copy cmd -: " << cpCmd;
        int rc = system(cpCmd.c_str());
        if (rc != 0) {
          edm::LogError("EvFDaqDirector") << " testmode: COPY EOL FAILED!!!!! -: " << cpCmd;
        }
      }

      return true;
    }
        BUEoLSFile = getEoLSFilePathOnBU(ls);
        eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
      }
    }
    // no new file found
    return false;
  }

void evf::EvFDaqDirector::createRunOpendirMaybe

(

)

Definition at line 760 of file EvFDaqDirector.cc.

References getRunOpenDirPath(), LogDebug, and cmsHarvester::path.

Referenced by EvFDaqDirector().

                                             {
    // create open dir if not already there

    boost::filesystem::path openPath = getRunOpenDirPath();
    if (!boost::filesystem::is_directory(openPath)) {
      LogDebug("EvFDaqDirector") << "<open> FU dir not found. Creating... -:" << openPath.string();
      boost::filesystem::create_directories(openPath);
    }
  }

std::string evf::EvFDaqDirector::eolsFileName ( const unsigned int  ls ) const

private

std::string evf::EvFDaqDirector::eorFileName ( ) const

private

std::string evf::EvFDaqDirector::findCurrentRunDir ( )

inline

Definition at line 56 of file EvFDaqDirector.h.

References dirManager_, evf::DirManager::findRunDir(), and run_.

{ return dirManager_.findRunDir(run_);}

std::string evf::EvFDaqDirector::getEoLSFilePathOnBU

(

const unsigned int

ls

)

const

Definition at line 360 of file EvFDaqDirector.cc.

References bu_run_dir_, fffnaming::eolsFileName(), and run_.

Referenced by bumpFile(), and FedRawDataInputSource::checkNextEvent().

                                                                           {
    return bu_run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
  }

std::string evf::EvFDaqDirector::getEoLSFilePathOnFU

(

const unsigned int

ls

)

const

Definition at line 364 of file EvFDaqDirector.cc.

References fffnaming::eolsFileName(), run_, and run_dir_.

Referenced by FedRawDataInputSource::checkNextEvent(), and FedRawDataInputSource::maybeOpenNewLumiSection().

                                                                           {
    return run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
  }

std::string evf::EvFDaqDirector::getEoRFilePath

(

)

const

Definition at line 368 of file EvFDaqDirector.cc.

References bu_run_dir_, fffnaming::eorFileName(), and run_.

Referenced by updateFuLock().

                                                 {
    return bu_run_dir_ + "/" + fffnaming::eorFileName(run_);
  }

std::string evf::EvFDaqDirector::getEoRFilePathOnFU

(

)

const

Definition at line 373 of file EvFDaqDirector.cc.

References fffnaming::eorFileName(), run_, and run_dir_.

Referenced by FedRawDataInputSource::checkNextEvent().

                                                     {
    return run_dir_ + "/" + fffnaming::eorFileName(run_);
  }

std::string evf::EvFDaqDirector::getInitFilePath

(

std::string const &

stream

)

const

Definition at line 332 of file EvFDaqDirector.cc.

References fffnaming::initFileNameWithPid(), run_, and run_dir_.

Referenced by DQMFileSaver::globalBeginRun().

                                                                         {
    return run_dir_ + "/" + fffnaming::initFileNameWithPid(run_,0,stream);
  }

std::string evf::EvFDaqDirector::getInputJsonFilePath	(	const unsigned int	ls,
		const unsigned int	index
	)		const

Definition at line 295 of file EvFDaqDirector.cc.

References bu_run_dir_, fffnaming::inputJsonFileName(), and run_.

Referenced by bumpFile().

                                                                                                      {
    return bu_run_dir_ + "/" + fffnaming::inputJsonFileName(run_,ls,index);
  }

std::string evf::EvFDaqDirector::getJumpFilePath ( ) const

inline

Definition at line 87 of file EvFDaqDirector.h.

References bu_run_dir_, getRunNumber(), fffnaming::inputRawFileName(), jumpIndex_, and jumpLS_.

Referenced by FedRawDataInputSource::renameToNextFree().

{ return bu_run_dir_ + "/" + fffnaming::inputRawFileName(getRunNumber(),jumpLS_,jumpIndex_); }

unsigned int evf::EvFDaqDirector::getJumpIndex ( ) const

inline

Definition at line 86 of file EvFDaqDirector.h.

References jumpIndex_.

{ return jumpIndex_; }

unsigned int evf::EvFDaqDirector::getJumpLS ( ) const

inline

Definition at line 85 of file EvFDaqDirector.h.

References jumpLS_.

{ return jumpLS_; }

std::string evf::EvFDaqDirector::getMergedDatChecksumFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 328 of file EvFDaqDirector.cc.

References hostname_, run_, run_dir_, and fffnaming::streamerDataChecksumFileNameWithInstance().

                                                                                                             {
    return run_dir_ + "/" + fffnaming::streamerDataChecksumFileNameWithInstance(run_,ls,stream,hostname_);
  }

std::string evf::EvFDaqDirector::getMergedDatFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 324 of file EvFDaqDirector.cc.

References hostname_, run_, run_dir_, and fffnaming::streamerDataFileNameWithInstance().

Referenced by maybeCreateAndLockFileHeadForStream().

                                                                                                     {
    return run_dir_ + "/" + fffnaming::streamerDataFileNameWithInstance(run_,ls,stream,hostname_);
  }

std::string evf::EvFDaqDirector::getMergedProtocolBufferHistogramFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 344 of file EvFDaqDirector.cc.

References hostname_, fffnaming::protocolBufferHistogramFileNameWithInstance(), run_, and run_dir_.

                                                                                                                         {
    return run_dir_ + "/" + fffnaming:: protocolBufferHistogramFileNameWithInstance(run_,ls,stream,hostname_);
  }

std::string evf::EvFDaqDirector::getMergedRootHistogramFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 356 of file EvFDaqDirector.cc.

References hostname_, fffnaming::rootHistogramFileNameWithInstance(), run_, and run_dir_.

                                                                                                               {
    return run_dir_ + "/" + fffnaming:: rootHistogramFileNameWithInstance(run_,ls,stream,hostname_);
  }

int evf::EvFDaqDirector::getNFilesFromEoLS ( std::string  BUEoLSFile )

private

Definition at line 524 of file EvFDaqDirector.cc.

References bu_run_dir_, data, def, jsoncollector::DataPoint::deserialize(), eolsNFilesIndex_, jsoncollector::DataPoint::getData(), jsoncollector::DataPoint::getDefinition(), jsoncollector::DataPointDefinition::getNames(), i, Json::Reader::parse(), readEolsDefinition_, matplotRender::reader, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by bumpFile().

                                                            {

    boost::filesystem::ifstream ij(BUEoLSFile);
    Json::Value deserializeRoot;
    Json::Reader reader;

    if (!reader.parse(ij, deserializeRoot)) {
      edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << BUEoLSFile;
      return -1;
    }

    std::string data;
    DataPoint dp;
    dp.deserialize(deserializeRoot);

    //read definition 
    if (readEolsDefinition_) {
      //std::string def = boost::algorithm::trim(dp.getDefinition());
      std::string def = dp.getDefinition();
      if (!def.size()) readEolsDefinition_=false;
      while (def.size()) {
        std::string fullpath;
        if (def.find('/')==0)
          fullpath = def;
        else
          fullpath = bu_run_dir_+'/'+def; 
        struct stat buf;
        if (stat(fullpath.c_str(), &buf) == 0) {
          DataPointDefinition eolsDpd;
          std::string defLabel = "legend";
          DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
          if (eolsDpd.getNames().size()==0) {
             //try with "data" label if "legend" format is not used
             eolsDpd = DataPointDefinition();
             defLabel="data";
             DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
          }
          for (unsigned int i=0;i<eolsDpd.getNames().size();i++)
            if (eolsDpd.getNames().at(i)=="NFiles")
              eolsNFilesIndex_ = i;
          readEolsDefinition_=false;
          break;
        }
        //check if we can still find definition
        if (def.size()<=1 || def.find('/')==std::string::npos) {
          readEolsDefinition_=false;
          break;
        }
        def = def.substr(def.find('/')+1);
      }
    }

    if (dp.getData().size()>eolsNFilesIndex_)
      data = dp.getData()[eolsNFilesIndex_];
    else {
      edm::LogError("EvFDaqDirector") << " error reading number of files from BU JSON -: " << BUEoLSFile;
      return -1;
    }
    return boost::lexical_cast<int>(data);
  }

std::string evf::EvFDaqDirector::getOpenDatFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 312 of file EvFDaqDirector.cc.

References run_, run_dir_, and fffnaming::streamerDataFileNameWithPid().

                                                                                                   {
    return run_dir_ + "/open/" + fffnaming::streamerDataFileNameWithPid(run_,ls,stream);
  }

std::string evf::EvFDaqDirector::getOpenInputJsonFilePath	(	const unsigned int	ls,
		const unsigned int	index
	)		const

Definition at line 308 of file EvFDaqDirector.cc.

References bu_run_dir_, fffnaming::inputJsonFileName(), and run_.

                                                                                                          {
    return bu_run_dir_ + "/open/" + fffnaming::inputJsonFileName(run_,ls,index);
  }

std::string evf::EvFDaqDirector::getOpenOutputJsonFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 316 of file EvFDaqDirector.cc.

References run_, run_dir_, and fffnaming::streamerJsonFileNameWithPid().

                                                                                                          {
    return run_dir_ + "/open/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
  }

std::string evf::EvFDaqDirector::getOpenProtocolBufferHistogramFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 336 of file EvFDaqDirector.cc.

References fffnaming::protocolBufferHistogramFileNameWithPid(), run_, and run_dir_.

                                                                                                                       {
    return run_dir_ + "/open/" + fffnaming::protocolBufferHistogramFileNameWithPid(run_,ls,stream);
  }

std::string evf::EvFDaqDirector::getOpenRawFilePath	(	const unsigned int	ls,
		const unsigned int	index
	)		const

Definition at line 304 of file EvFDaqDirector.cc.

References bu_run_dir_, fffnaming::inputRawFileName(), and run_.

                                                                                                    {
    return bu_run_dir_ + "/open/" + fffnaming::inputRawFileName(run_,ls,index);
  }

std::string evf::EvFDaqDirector::getOpenRootHistogramFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 348 of file EvFDaqDirector.cc.

References fffnaming::rootHistogramFileNameWithPid(), run_, and run_dir_.

                                                                                                             {
    return run_dir_ + "/open/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
  }

std::string evf::EvFDaqDirector::getOutputJsonFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 320 of file EvFDaqDirector.cc.

References run_, run_dir_, and fffnaming::streamerJsonFileNameWithPid().

                                                                                                      {
    return run_dir_ + "/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
  }

std::string evf::EvFDaqDirector::getProtocolBufferHistogramFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 340 of file EvFDaqDirector.cc.

References fffnaming::protocolBufferHistogramFileNameWithPid(), run_, and run_dir_.

                                                                                                                   {
    return run_dir_ + "/" + fffnaming::protocolBufferHistogramFileNameWithPid(run_,ls,stream);
  }

std::string evf::EvFDaqDirector::getRawFilePath	(	const unsigned int	ls,
		const unsigned int	index
	)		const

Definition at line 300 of file EvFDaqDirector.cc.

References bu_run_dir_, fffnaming::inputRawFileName(), and run_.

Referenced by removeFile().

                                                                                                {
    return bu_run_dir_ + "/" + fffnaming::inputRawFileName(run_,ls,index);
  }

std::string evf::EvFDaqDirector::getRootHistogramFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 352 of file EvFDaqDirector.cc.

References fffnaming::rootHistogramFileNameWithPid(), run_, and run_dir_.

                                                                                                         {
    return run_dir_ + "/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
  }

unsigned int evf::EvFDaqDirector::getRunNumber ( ) const

inline

Definition at line 84 of file EvFDaqDirector.h.

References run_.

Referenced by getJumpFilePath().

{ return run_; }

std::string evf::EvFDaqDirector::getRunOpenDirPath ( ) const

inline

Definition at line 77 of file EvFDaqDirector.h.

References run_dir_.

Referenced by createRunOpendirMaybe(), and EvFDaqDirector().

{return run_dir_ +"/open";}

std::vector<int>* evf::EvFDaqDirector::getStreamFileTracker ( )

inline

Definition at line 95 of file EvFDaqDirector.h.

References streamFileTracker_.

Referenced by FedRawDataInputSource::getNextEvent().

{return &streamFileTracker_;}

bool evf::EvFDaqDirector::getTestModeNoBuilderUnit ( )

inline

Definition at line 88 of file EvFDaqDirector.h.

References testModeNoBuilderUnit_.

{ return testModeNoBuilderUnit_;}

std::string evf::EvFDaqDirector::initFileName ( std::string const &  stream ) const

private

std::string evf::EvFDaqDirector::inputFileNameStem ( const unsigned int  ls, const unsigned int  index  ) const

private

bool evf::EvFDaqDirector::isSingleStreamThread ( )

inline

Definition at line 96 of file EvFDaqDirector.h.

References nStreams_, and nThreads_.

Referenced by FedRawDataInputSource::getNextEvent().

{return nStreams_==1 && nThreads_==1;}

void evf::EvFDaqDirector::lockFULocal

(

)

Definition at line 738 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by updateFuLock().

                                   {
    //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_flk);
    flock(fulocal_rwlock_fd_,LOCK_EX);
  }

void evf::EvFDaqDirector::lockFULocal2

(

)

Definition at line 749 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

Referenced by FedRawDataInputSource::checkNextEvent(), and FedRawDataInputSource::maybeOpenNewLumiSection().

                                    {
    //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_flk2);
    flock(fulocal_rwlock_fd2_,LOCK_EX);
  }

void evf::EvFDaqDirector::lockInitLock

(

)

Definition at line 730 of file EvFDaqDirector.cc.

References init_lock_.

                                    {
    pthread_mutex_lock(&init_lock_);
  }

FILE * evf::EvFDaqDirector::maybeCreateAndLockFileHeadForStream	(	unsigned int	ls,
		std::string &	stream
	)

Definition at line 710 of file EvFDaqDirector.cc.

References data_readwrite_fd_, data_rw_flk, data_rw_stream, getMergedDatFilePath(), and LogDebug.

                                                                                              {
    data_rw_stream = fopen(getMergedDatFilePath(ls,stream).c_str(), "a"); //open stream for appending
    data_readwrite_fd_ = fileno(data_rw_stream);
    if (data_readwrite_fd_ == -1)
      edm::LogError("EvFDaqDirector") << "problem with creating filedesc for datamerge "
        << strerror(errno);
    else
      LogDebug("EvFDaqDirector") << "creating filedesc for datamerge -: "
        << data_readwrite_fd_;
    fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_flk);

    return data_rw_stream;
  }

std::string evf::EvFDaqDirector::mergedFileNameStem ( const unsigned int  ls, std::string const &  stream  ) const

private

void evf::EvFDaqDirector::openFULockfileStream ( std::string &  fuLockFilePath, bool  create  )

private

Definition at line 691 of file EvFDaqDirector.cc.

References fu_readwritelock_fd_, fu_rw_lock_stream, and LogDebug.

Referenced by EvFDaqDirector().

                                                                              {
    if (create) {
      fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
                  S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
      chmod(fulockfile.c_str(),0766);
    } else {
      fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
    }
    if (fu_readwritelock_fd_ == -1)
      edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
                                      << " create:" << create << " error:" << strerror(errno);
    else
      LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: "
        << fu_readwritelock_fd_;

    fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
  }

bool evf::EvFDaqDirector::outputAdler32Recheck ( ) const

inline

Definition at line 78 of file EvFDaqDirector.h.

References outputAdler32Recheck_.

{return outputAdler32Recheck_;}

std::string evf::EvFDaqDirector::outputFileNameStem ( const unsigned int  ls, std::string const &  stream  ) const

private

void evf::EvFDaqDirector::postEndRun

(

edm::GlobalContext const &

globalContext

)

Definition at line 216 of file EvFDaqDirector.cc.

References bu_readlock_fd_, bu_run_dir_, bu_writelock_fd_, directorBu_, lut2db_cfg::filename, removeFile(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by EvFDaqDirector().

                                                                       {
    close(bu_readlock_fd_);
    close(bu_writelock_fd_);
    if (directorBu_) {
      std::string filename = bu_run_dir_ + "/bu.lock";
      removeFile(filename);
    }
  }

void evf::EvFDaqDirector::preallocate

(

edm::service::SystemBounds const &

bounds

)

Definition at line 225 of file EvFDaqDirector.cc.

References i, edm::service::SystemBounds::maxNumberOfStreams(), edm::service::SystemBounds::maxNumberOfThreads(), nStreams_, nThreads_, and streamFileTracker_.

Referenced by EvFDaqDirector().

                                                                       {

    for (unsigned int i=0;i<bounds.maxNumberOfStreams();i++){
      streamFileTracker_.push_back(-1);
    }
    nThreads_=bounds.maxNumberOfStreams();
    nStreams_=bounds.maxNumberOfThreads();
  }

void evf::EvFDaqDirector::preBeginRun

(

edm::GlobalContext const &

globalContext

)

Definition at line 234 of file EvFDaqDirector.cc.

References dirManager_, evf::DirManager::findHighestRunDir(), and run_dir_.

Referenced by EvFDaqDirector().

                                                                        {

//    assert(run_ == id.run());

    // check if the requested run is the latest one - issue a warning if it isn't
    if (dirManager_.findHighestRunDir() != run_dir_) {
      edm::LogWarning("EvFDaqDirector") << "WARNING - checking run dir -: "
                    << run_dir_ << ". This is not the highest run "
                    << dirManager_.findHighestRunDir();
    }
  }

void evf::EvFDaqDirector::preGlobalEndLumi

(

edm::GlobalContext const &

globalContext

)

Definition at line 246 of file EvFDaqDirector.cc.

References cppFunctionSkipper::exception, fileDeleteLockPtr_, filesToDeletePtr_, LogDebug, python.rootplot.utilities::ls(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), cmsHarvester::path, and python.multivaluedict::remove().

Referenced by EvFDaqDirector().

  {
    //delete all files belonging to just closed lumi
    unsigned int ls = globalContext.luminosityBlockID().luminosityBlock();
    if (!fileDeleteLockPtr_ || !filesToDeletePtr_) {
      edm::LogWarning("EvFDaqDirector") << " Handles to check for files to delete were not set by the input source...";
      return;
    }

    std::unique_lock<std::mutex> lkw(*fileDeleteLockPtr_);
    auto it = filesToDeletePtr_->begin();
    while (it!=filesToDeletePtr_->end()) {
      if (it->second->lumi_ ==  ls) {
        const boost::filesystem::path filePath(it->second->fileName_);
        LogDebug("EvFDaqDirector") << "Deleting input file -:" << it->second->fileName_;
        try {
          //rarely this fails but file gets deleted
          boost::filesystem::remove(filePath);
        }
        catch (const boost::filesystem::filesystem_error& ex)
        {
          edm::LogError("EvFDaqDirector") << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() << ". Trying again.";
          usleep(10000);
          try {
            boost::filesystem::remove(filePath);
          }
            catch (...) {/*file gets deleted first time but exception is still thrown*/}
        }
        catch (std::exception& ex)
        {
          edm::LogError("EvFDaqDirector") << " - deleteFile std::exception CAUGHT -: " << ex.what() << ". Trying again.";
          usleep(10000);
          try {
        boost::filesystem::remove(filePath);
          } catch (...) {/*file gets deleted first time but exception is still thrown*/}
        }
        
        delete it->second;
    it = filesToDeletePtr_->erase(it);
      }
      else it++;
    }
  }

void evf::EvFDaqDirector::preSourceEvent ( edm::StreamID const &  streamID )

inline

Definition at line 290 of file EvFDaqDirector.cc.

References currentFileIndex_, and streamFileTracker_.

Referenced by EvFDaqDirector().

                                                                        {
    streamFileTracker_[streamID]=currentFileIndex_;
  }

int evf::EvFDaqDirector::readLastLSEntry

(

std::string const &

file

)

Definition at line 771 of file EvFDaqDirector.cc.

References Json::Value::asInt(), mergeVDriftHistosByStation::file, Json::Value::get(), Json::Reader::parse(), matplotRender::reader, and run_regression::ret.

Referenced by updateFuLock().

                                                           {

    boost::filesystem::ifstream ij(file);
    Json::Value deserializeRoot;
    Json::Reader reader;

    if (!reader.parse(ij, deserializeRoot)) {
      edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << file;
      return -1;
    }

    int ret = deserializeRoot.get("lastLS","").asInt();
    return ret;

  }

void evf::EvFDaqDirector::removeFile	(	unsigned int	ls,
		unsigned int	index
	)

Definition at line 384 of file EvFDaqDirector.cc.

References getRawFilePath().

Referenced by postEndRun().

                                                                     {
    removeFile(getRawFilePath(ls,index));
  }

void evf::EvFDaqDirector::removeFile

(

std::string

filename

)

Definition at line 377 of file EvFDaqDirector.cc.

                                                    {
    int retval = remove(filename.c_str());
    if (retval != 0)
      edm::LogError("EvFDaqDirector") << "Could not remove used file -: " << filename << ". error = "
                              << strerror(errno);
  }

void evf::EvFDaqDirector::setDeleteTracking ( std::mutex *  fileDeleteLock, std::list< std::pair< int, InputFile * >> *  filesToDelete  )

inline

Definition at line 103 of file EvFDaqDirector.h.

References fileDeleteLockPtr_, and filesToDeletePtr_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

                                                                                                        {
        fileDeleteLockPtr_=fileDeleteLock;
        filesToDeletePtr_ = filesToDelete;
      }

void evf::EvFDaqDirector::setFMS ( evf::FastMonitoringService *  fms )

inline

Definition at line 93 of file EvFDaqDirector.h.

References fms_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

{fms_=fms;}

void evf::EvFDaqDirector::tryInitializeFuLockFile

(

)

Definition at line 676 of file EvFDaqDirector.cc.

References fu_rw_lock_stream, getHLTprescales::readIndex(), and testModeNoBuilderUnit_.

Referenced by EvFDaqDirector().

                                               {
    if (fu_rw_lock_stream == 0)
      edm::LogError("EvFDaqDirector") << "Error creating fu read/write lock stream "
                      << strerror(errno);
    else {
      edm::LogInfo("EvFDaqDirector") << "Initializing FU LOCK FILE";
      unsigned int readLs = 1, readIndex = 0, jumpLs = 3, jumpIndex = 0;
      if (testModeNoBuilderUnit_)
    fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs, readIndex,
        jumpLs, jumpIndex);
      else
    fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
    }
  }

void evf::EvFDaqDirector::unlockAndCloseMergeStream

(

)

Definition at line 724 of file EvFDaqDirector.cc.

References data_readwrite_fd_, data_rw_fulk, and data_rw_stream.

                                                 {
    fflush(data_rw_stream);
    fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_fulk);
    fclose(data_rw_stream);
  }

void evf::EvFDaqDirector::unlockFULocal

(

)

Definition at line 743 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by FedRawDataInputSource::grabNextJsonFile(), and ~EvFDaqDirector().

                                     {
    //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_fulk);
    flock(fulocal_rwlock_fd_,LOCK_UN);
  }

void evf::EvFDaqDirector::unlockFULocal2

(

)

Definition at line 754 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

Referenced by FedRawDataInputSource::maybeOpenNewLumiSection(), and ~EvFDaqDirector().

                                      {
    //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_fulk2);
    flock(fulocal_rwlock_fd2_,LOCK_UN);
  }

void evf::EvFDaqDirector::unlockInitLock

(

)

Definition at line 734 of file EvFDaqDirector.cc.

References init_lock_.

                                      {
    pthread_mutex_unlock(&init_lock_);
  }

void evf::EvFDaqDirector::updateFileIndex ( int const &  fileIndex )

inline

Definition at line 94 of file EvFDaqDirector.h.

References currentFileIndex_.

Referenced by FedRawDataInputSource::getNextEvent().

{currentFileIndex_=fileIndex;}

EvFDaqDirector::FileStatus evf::EvFDaqDirector::updateFuLock	(	unsigned int &	ls,
		std::string &	nextFile,
		uint32_t &	fsize
	)

Definition at line 388 of file EvFDaqDirector.cc.

References bu_run_dir_, bumpFile(), DDVectorGetter::check(), fu_readwritelock_fd_, fu_rw_flk, fu_rw_fulk, fu_rw_lock_stream, getEoRFilePath(), jumpIndex_, jumpLS_, lockFULocal(), LogDebug, newFile, noFile, getHLTprescales::readIndex(), readLastLSEntry(), runEnded, stopFilePath_, and testModeNoBuilderUnit_.

Referenced by FedRawDataInputSource::readSupervisor().

                                                                                                              {
    EvFDaqDirector::FileStatus fileStatus = noFile;

    int retval = -1;
    int lock_attempts = 0;

    struct stat buf;
    int stopFileLS = -1;
    if (stat(stopFilePath_.c_str(),&buf)==0) {
        stopFileLS = readLastLSEntry(stopFilePath_);
        edm::LogWarning("EvFDaqDirector") << "Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
        //return runEnded;
    }

    while (retval==-1) {
      retval = fcntl(fu_readwritelock_fd_, F_SETLK, &fu_rw_flk);
      if (retval==-1) usleep(50000);
      else continue;

      lock_attempts++;
      if (lock_attempts>100 ||  errno==116) {
        if (errno==116)
          edm::LogWarning("EvFDaqDirector") << "Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
        else
          edm::LogWarning("EvFDaqDirector") << "Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno "<< errno <<":"<< strerror(errno) << std::endl;

        if (stat(bu_run_dir_.c_str(), &buf)!=0) return runEnded;
        if (stat((bu_run_dir_+"/fu.lock").c_str(), &buf)!=0) return runEnded;
        lock_attempts=0;
      }
    }
    if(retval!=0) return fileStatus;

#ifdef DEBUG
    timeval ts_lockend;
    gettimeofday(&ts_lockend,0);
#endif

    // if the stream is readable
    if (fu_rw_lock_stream != 0) {
      unsigned int readLs, readIndex, jumpLs, jumpIndex;
      int check = 0;
      // rewind the stream
      check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
      // if rewinded ok
      if (check == 0) {
    // read its' values
    if (testModeNoBuilderUnit_)
      fscanf(fu_rw_lock_stream, "%u %u %u %u", &readLs, &readIndex,
         &jumpLs, &jumpIndex);
    else {
      fscanf(fu_rw_lock_stream, "%u %u", &readLs, &readIndex);
      edm::LogInfo("EvFDaqDirector") << "Read fu.lock file file -: " << readLs << ":" << readIndex;
        }

    // try to bump
    bool bumpedOk = bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
    ls = readLs;
    // there is a new file to grab or lumisection ended
    if (bumpedOk) {
      // rewind and clear
      check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
      if (check == 0) {
        ftruncate(fu_readwritelock_fd_, 0);
        fflush(fu_rw_lock_stream); //this should not be needed ???
      } else
          edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
                                          << strerror(errno);
      // write new data
      check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
      if (check == 0) {
        // write next index in the file, which is the file the next process should take
        if (testModeNoBuilderUnit_) {
          fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs,
              readIndex + 1, readLs + 2, readIndex + 1);
          jumpLS_ = readLs + 2;
          jumpIndex_ = readIndex;
        } else {
          fprintf(fu_rw_lock_stream, "%u %u", readLs,
              readIndex + 1);
        }
        fflush(fu_rw_lock_stream);
        fsync(fu_readwritelock_fd_);

        fileStatus = newFile;

        if (testModeNoBuilderUnit_)
          edm::LogInfo("EvFDaqDirector") << "Written to file -: " << readLs << ":"
                                 << readIndex + 1 << " --> " << readLs + 2
                                 << ":" << readIndex + 1;
        else
          LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":"
                                 << readIndex + 1;

      } else
          edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
                                          << strerror(errno);
    }
      } else
    edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for reading failed with error "
                    << strerror(errno);
    } else
      edm::LogError("EvFDaqDirector") << "fu read/write lock stream is invalid " << strerror(errno);

#ifdef DEBUG
    timeval ts_preunlock;
    gettimeofday(&ts_preunlock,0);
    int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
    double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
#endif

    //if new json is present, lock file which FedRawDataInputSource will later unlock
    if (fileStatus==newFile) lockFULocal();

    //release lock at this point
    int retvalu=-1;
    retvalu=fcntl(fu_readwritelock_fd_, F_SETLKW, &fu_rw_fulk);
    if (retvalu==-1) edm::LogError("EvFDaqDirector") << "Error unlocking the fu.lock " << strerror(errno);

#ifdef DEBUG
    edm::LogDebug("EvFDaqDirector") << "Waited during lock -: " << locked_period << " seconds";
#endif

    if ( fileStatus == noFile ) {
      struct stat buf;
      //edm::LogInfo("EvFDaqDirector") << " looking for EoR file: " << getEoRFilePath().c_str();
      if ( stat(getEoRFilePath().c_str(), &buf) == 0 || stat(bu_run_dir_.c_str(), &buf)!=0)
        fileStatus = runEnded;
      if (stopFileLS>=0 && (int)ls > stopFileLS) {
         edm::LogInfo("EvFDaqDirector") << "Reached maximum lumisection set by hltd";
         fileStatus = runEnded;
      }
    }
    return fileStatus;
  }

Member Data Documentation

std::string evf::EvFDaqDirector::base_dir_

private

Definition at line 123 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

std::string evf::EvFDaqDirector::bu_base_dir_

private

Definition at line 124 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

struct flock evf::EvFDaqDirector::bu_r_flk

private

Definition at line 155 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_r_fulk

private

Definition at line 157 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_r_lock_stream

private

Definition at line 143 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_readlock_fd_

private

Definition at line 135 of file EvFDaqDirector.h.

Referenced by postEndRun().

std::string evf::EvFDaqDirector::bu_run_dir_

private

Definition at line 132 of file EvFDaqDirector.h.

Referenced by buBaseRunDir(), EvFDaqDirector(), getEoLSFilePathOnBU(), getEoRFilePath(), getInputJsonFilePath(), getJumpFilePath(), getNFilesFromEoLS(), getOpenInputJsonFilePath(), getOpenRawFilePath(), getRawFilePath(), postEndRun(), and updateFuLock().

std::string evf::EvFDaqDirector::bu_run_open_dir_

private

Definition at line 133 of file EvFDaqDirector.h.

Referenced by buBaseRunOpenDir(), and EvFDaqDirector().

FILE* evf::EvFDaqDirector::bu_t_monitor_stream

private

Definition at line 146 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_flk

private

Definition at line 154 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_fulk

private

Definition at line 156 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_w_lock_stream

private

Definition at line 142 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

FILE* evf::EvFDaqDirector::bu_w_monitor_stream

private

Definition at line 145 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_writelock_fd_

private

Definition at line 136 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and postEndRun().

int evf::EvFDaqDirector::currentFileIndex_ = -1

private

Definition at line 169 of file EvFDaqDirector.h.

Referenced by preSourceEvent(), and updateFileIndex().

int evf::EvFDaqDirector::data_readwrite_fd_

private

Definition at line 138 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream(), and unlockAndCloseMergeStream().

struct flock evf::EvFDaqDirector::data_rw_flk

private

Definition at line 160 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream().

struct flock evf::EvFDaqDirector::data_rw_fulk

private

Definition at line 161 of file EvFDaqDirector.h.

Referenced by unlockAndCloseMergeStream().

FILE* evf::EvFDaqDirector::data_rw_stream

private

Definition at line 147 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream(), and unlockAndCloseMergeStream().

bool evf::EvFDaqDirector::directorBu_

private

Definition at line 125 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and postEndRun().

DirManager evf::EvFDaqDirector::dirManager_

private

Definition at line 149 of file EvFDaqDirector.h.

Referenced by findCurrentRunDir(), and preBeginRun().

unsigned int evf::EvFDaqDirector::eolsNFilesIndex_ = 1

private

Definition at line 180 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

std::mutex* evf::EvFDaqDirector::fileDeleteLockPtr_ = nullptr

private

Definition at line 171 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

std::list<std::pair<int,InputFile*> >* evf::EvFDaqDirector::filesToDeletePtr_ = nullptr

private

Definition at line 172 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

evf::FastMonitoringService* evf::EvFDaqDirector::fms_ = nullptr

private

Definition at line 167 of file EvFDaqDirector.h.

Referenced by bumpFile(), and setFMS().

int evf::EvFDaqDirector::fu_readwritelock_fd_

private

Definition at line 137 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), openFULockfileStream(), and updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_flk

private

Definition at line 158 of file EvFDaqDirector.h.

Referenced by updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_fulk

private

Definition at line 159 of file EvFDaqDirector.h.

Referenced by updateFuLock().

FILE* evf::EvFDaqDirector::fu_rw_lock_stream

private

Definition at line 144 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), openFULockfileStream(), tryInitializeFuLockFile(), and updateFuLock().

int evf::EvFDaqDirector::fulocal_rwlock_fd2_

private

Definition at line 140 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), lockFULocal2(), unlockFULocal2(), and ~EvFDaqDirector().

int evf::EvFDaqDirector::fulocal_rwlock_fd_

private

Definition at line 139 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), lockFULocal(), unlockFULocal(), and ~EvFDaqDirector().

std::string evf::EvFDaqDirector::hostname_

private

Definition at line 129 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), getMergedDatChecksumFilePath(), getMergedDatFilePath(), getMergedProtocolBufferHistogramFilePath(), and getMergedRootHistogramFilePath().

pthread_mutex_t evf::EvFDaqDirector::init_lock_ = PTHREAD_MUTEX_INITIALIZER

private

Definition at line 174 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), lockInitLock(), and unlockInitLock().

unsigned int evf::EvFDaqDirector::jumpIndex_

private

Definition at line 152 of file EvFDaqDirector.h.

Referenced by getJumpFilePath(), getJumpIndex(), and updateFuLock().

unsigned int evf::EvFDaqDirector::jumpLS_

private

Definition at line 152 of file EvFDaqDirector.h.

Referenced by getJumpFilePath(), getJumpLS(), and updateFuLock().

unsigned int evf::EvFDaqDirector::nStreams_ =0

private

Definition at line 176 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

unsigned int evf::EvFDaqDirector::nThreads_ =0

private

Definition at line 177 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

bool evf::EvFDaqDirector::outputAdler32Recheck_

private

Definition at line 127 of file EvFDaqDirector.h.

Referenced by outputAdler32Recheck().

unsigned long evf::EvFDaqDirector::previousFileSize_

private

Definition at line 151 of file EvFDaqDirector.h.

Referenced by bumpFile().

bool evf::EvFDaqDirector::readEolsDefinition_ = true

private

Definition at line 179 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

unsigned int evf::EvFDaqDirector::run_

private

Definition at line 126 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), findCurrentRunDir(), getEoLSFilePathOnBU(), getEoLSFilePathOnFU(), getEoRFilePath(), getEoRFilePathOnFU(), getInitFilePath(), getInputJsonFilePath(), getMergedDatChecksumFilePath(), getMergedDatFilePath(), getMergedProtocolBufferHistogramFilePath(), getMergedRootHistogramFilePath(), getOpenDatFilePath(), getOpenInputJsonFilePath(), getOpenOutputJsonFilePath(), getOpenProtocolBufferHistogramFilePath(), getOpenRawFilePath(), getOpenRootHistogramFilePath(), getOutputJsonFilePath(), getProtocolBufferHistogramFilePath(), getRawFilePath(), getRootHistogramFilePath(), and getRunNumber().

std::string evf::EvFDaqDirector::run_dir_

private

Definition at line 131 of file EvFDaqDirector.h.

Referenced by baseRunDir(), EvFDaqDirector(), getEoLSFilePathOnFU(), getEoRFilePathOnFU(), getInitFilePath(), getMergedDatChecksumFilePath(), getMergedDatFilePath(), getMergedProtocolBufferHistogramFilePath(), getMergedRootHistogramFilePath(), getOpenDatFilePath(), getOpenOutputJsonFilePath(), getOpenProtocolBufferHistogramFilePath(), getOpenRootHistogramFilePath(), getOutputJsonFilePath(), getProtocolBufferHistogramFilePath(), getRootHistogramFilePath(), getRunOpenDirPath(), and preBeginRun().

std::string evf::EvFDaqDirector::run_string_

private

Definition at line 130 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

std::string evf::EvFDaqDirector::stopFilePath_

private

Definition at line 181 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateFuLock().

std::vector<int> evf::EvFDaqDirector::streamFileTracker_

private

Definition at line 168 of file EvFDaqDirector.h.

Referenced by getStreamFileTracker(), preallocate(), and preSourceEvent().

bool evf::EvFDaqDirector::testModeNoBuilderUnit_

private

Definition at line 122 of file EvFDaqDirector.h.

Referenced by bumpFile(), getTestModeNoBuilderUnit(), tryInitializeFuLockFile(), and updateFuLock().