#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	checkMergeTypePSet (edm::ProcessContext const &pc)

void	checkTransferSystemPSet (edm::ProcessContext const &pc)

void	createProcessingNotificationMaybe () const

void	createRunOpendirMaybe ()

bool	emptyLumisectionMode () const

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

std::string	findCurrentRunDir ()

std::string	getBoLSFilePathOnFU (const unsigned int ls) const

std::string	getDatFilePath (const unsigned int ls, std::string const &stream) const

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	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	getOpenInitFilePath (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::string	getStreamDestinations (std::string const &stream) const

std::vector< int > *	getStreamFileTracker ()

std::string	getStreamMergeType (std::string const &stream, MergeType defaultType)

void	initRun ()

bool	isSingleStreamThread ()

void	lockFULocal ()

void	lockFULocal2 ()

void	lockInitLock ()

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

bool	microMergeDisabled () const

bool	outputAdler32Recheck () const

void	overrideRunNumber (unsigned int run)

void	postEndRun (edm::GlobalContext const &globalContext)

void	preallocate (edm::service::SystemBounds const &bounds)

void	preBeginJob (edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &)

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, uint64_t &lockWaitTime)

	~EvFDaqDirector ()

Static Public Member Functions
static void	fillDescriptions (edm::ConfigurationDescriptions &descriptions)

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_

bool	emptyLumisectionMode_

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_

unsigned int	fuLockPollInterval_

std::string	hltSourceDirectory_

std::string	hostname_

pthread_mutex_t	init_lock_ = PTHREAD_MUTEX_INITIALIZER

std::map< std::string, std::string >	mergeTypeMap_

std::string	mergeTypePset_

bool	microMergeDisabled_

unsigned int	nStreams_ =0

unsigned int	nThreads_ =0

bool	outputAdler32Recheck_

unsigned long	previousFileSize_

bool	readEolsDefinition_ = true

bool	requireTSPSet_

unsigned int	run_

std::string	run_dir_

std::string	run_string_

std::string	selectedTransferMode_

unsigned int	stop_ls_override_ = 0

std::string	stopFilePath_

std::string	stopFilePathPid_

std::vector< int >	streamFileTracker_

std::shared_ptr< Json::Value >	transferSystemJson_

Static Private Attributes
static const std::vector < std::string >	MergeTypeNames_ = {"","DAT","PB","JSNDATA"}

Detailed Description

Definition at line 54 of file EvFDaqDirector.h.

Member Enumeration Documentation

enum evf::EvFDaqDirector::FileStatus

Enumerator
noFile
sameFile
newFile
newLumi
runEnded
runAbort

Definition at line 58 of file EvFDaqDirector.h.

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

evf::EvFDaqDirector::runAbort

Definition: EvFDaqDirector.h:58

evf::EvFDaqDirector::sameFile

Definition: EvFDaqDirector.h:58

evf::EvFDaqDirector::newLumi

Definition: EvFDaqDirector.h:58

evf::EvFDaqDirector::noFile

Definition: EvFDaqDirector.h:58

evf::EvFDaqDirector::newFile

Definition: EvFDaqDirector.h:58

evf::EvFDaqDirector::runEnded

Definition: EvFDaqDirector.h:58

Constructor & Destructor Documentation

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

explicit

Definition at line 47 of file EvFDaqDirector.cc.

References emptyLumisectionMode_, fuLockPollInterval_, hostname_, microMergeDisabled_, postEndRun(), preallocate(), preBeginJob(), preBeginRun(), preGlobalEndLumi(), preSourceEvent(), AlCaHLTBitMon_QueryRunRegistry::string, edm::ActivityRegistry::watchPostGlobalEndRun(), edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreBeginJob(), edm::ActivityRegistry::watchPreGlobalBeginRun(), edm::ActivityRegistry::watchPreGlobalEndLumi(), and edm::ActivityRegistry::watchPreSourceEvent().

                                            :
     base_dir_(pset.getUntrackedParameter<std::string> ("baseDir", ".")),
     bu_base_dir_(pset.getUntrackedParameter<std::string> ("buBaseDir", ".")),
     directorBu_(pset.getUntrackedParameter<bool> ("directorIsBu", false)),
     run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
     outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck",false)),
     requireTSPSet_(pset.getUntrackedParameter<bool>("requireTransfersPSet",false)),
     selectedTransferMode_(pset.getUntrackedParameter<std::string>("selectedTransferMode","")),
     hltSourceDirectory_(pset.getUntrackedParameter<std::string>("hltSourceDirectory","")),
     fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>("fuLockPollInterval",2000)),
     emptyLumisectionMode_(pset.getUntrackedParameter<bool>("emptyLumisectionMode",true)),
     microMergeDisabled_(pset.getUntrackedParameter<bool>("microMergeDisabled",true)),
     mergeTypePset_(pset.getUntrackedParameter<std::string>("mergeTypePset","")),
     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),
 
     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.watchPreBeginJob(this, &EvFDaqDirector::preBeginJob);
     reg.watchPreGlobalBeginRun(this, &EvFDaqDirector::preBeginRun);
     reg.watchPostGlobalEndRun(this, &EvFDaqDirector::postEndRun);
     reg.watchPreSourceEvent(this, &EvFDaqDirector::preSourceEvent);
     reg.watchPreGlobalEndLumi(this,&EvFDaqDirector::preGlobalEndLumi);
 
     //save hostname for later
     char hostname[33];
     gethostname(hostname,32);
     hostname_ = hostname;
 
     char * fuLockPollIntervalPtr = getenv("FFF_LOCKPOLLINTERVAL");
     if (fuLockPollIntervalPtr) {
       try {
         fuLockPollInterval_=boost::lexical_cast<unsigned int>(std::string(fuLockPollIntervalPtr));
         edm::LogInfo("EvFDaqDirector") << "Setting fu lock poll interval by environment string: " << fuLockPollInterval_ << " us";
       }
       catch( boost::bad_lexical_cast const& ) {
         edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fuLockPollIntervalPtr);
       }
     }
 
     char * emptyLumiModePtr = getenv("FFF_EMPTYLSMODE");
     if (emptyLumiModePtr) {
         emptyLumisectionMode_ = true;
         edm::LogInfo("EvFDaqDirector") << "Setting empty lumisection mode";
     }
 
     char * microMergeDisabledPtr = getenv("FFF_MICROMERGEDISABLED");
     if (microMergeDisabledPtr) {
         microMergeDisabled_ = true;
         edm::LogInfo("EvFDaqDirector") << "Disabling dat file micro-merge by the HLT process (delegated to the hlt daemon)";
     }
   }

evf::EvFDaqDirector::~EvFDaqDirector ( )

Definition at line 260 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 72 of file EvFDaqDirector.h.

References run_dir_.

72 {return run_dir_;}

evf::EvFDaqDirector::run_dir_

std::string run_dir_

Definition: EvFDaqDirector.h:163

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

inline

Definition at line 73 of file EvFDaqDirector.h.

References bu_run_dir_.

73 {return bu_run_dir_;}

evf::EvFDaqDirector::bu_run_dir_

std::string bu_run_dir_

Definition: EvFDaqDirector.h:164

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

inline

Definition at line 74 of file EvFDaqDirector.h.

References bu_run_open_dir_.

74 {return bu_run_open_dir_;}

evf::EvFDaqDirector::bu_run_open_dir_

std::string bu_run_open_dir_

Definition: EvFDaqDirector.h:165

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

private

Definition at line 726 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

                                                                                                                       {
 
     if (previousFileSize_ != 0) {
       if (!fms_) {
           fms_ = (FastMonitoringService *) (edm::Service<evf::MicroStateService>().operator->());
       }
       if (fms_) fms_->accumulateFileSize(ls, previousFileSize_);
       previousFileSize_ = 0;
     }
 
     //reached limit
     if (maxLS>=0 && ls > (unsigned int)maxLS) return false; 
 
     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);
       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;
       return true;
     }
         else {
           //change of policy: we need to cycle through each LS
           return false;
         }
         BUEoLSFile = getEoLSFilePathOnBU(ls);
         eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
       }
     }
     // no new file found
     return false;
   }

void evf::EvFDaqDirector::checkMergeTypePSet ( edm::ProcessContext const & pc )

Definition at line 1015 of file EvFDaqDirector.cc.

References edm::ParameterSet::existsAs(), edm::ParameterSet::getParameterSet(), edm::getParameterSet(), mergeTypeMap_, mergeTypePset_, edm::ProcessContext::parameterSetID(), unpackData-CaloStage2::pname, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by preBeginJob().

   {
     if (mergeTypePset_.empty()) return;
     if(mergeTypeMap_.size()) return;
     edm::ParameterSet const& topPset = edm::getParameterSet(pc.parameterSetID());
     if (topPset.existsAs<edm::ParameterSet>(mergeTypePset_,true))
     {
       const edm::ParameterSet& tsPset(topPset.getParameterSet(mergeTypePset_));
       for (std::string pname : tsPset.getParameterNames()) {
           std::string streamType = tsPset.getParameter<std::string>(pname); 
           mergeTypeMap_[pname]=streamType;
       }
     }
   }

void evf::EvFDaqDirector::checkTransferSystemPSet ( edm::ProcessContext const & pc )

Definition at line 912 of file EvFDaqDirector.cc.

References Json::Value::append(), Json::arrayValue, alignCSCRings::dest, myMessageLogger_cff::destinations, Exception, edm::ParameterSet::existsAs(), edm::ParameterSet::getParameter(), edm::ParameterSet::getParameterSet(), edm::getParameterSet(), universalConfigTemplate::mode, edm::ProcessContext::parameterSetID(), requireTSPSet_, and transferSystemJson_.

Referenced by preBeginJob().

   {
     if(transferSystemJson_) return;
 
     transferSystemJson_.reset(new Json::Value);
     edm::ParameterSet const& topPset = edm::getParameterSet(pc.parameterSetID());
     if (topPset.existsAs<edm::ParameterSet>("transferSystem",true))
     {
       const edm::ParameterSet& tsPset(topPset.getParameterSet("transferSystem"));
 
       Json::Value destinationsVal(Json::arrayValue);
       std::vector<std::string> destinations = tsPset.getParameter<std::vector<std::string>>("destinations");
       for (auto & dest: destinations) destinationsVal.append(dest);
       (*transferSystemJson_)["destinations"]=destinationsVal;
 
       Json::Value modesVal(Json::arrayValue);
       std::vector<std::string> modes = tsPset.getParameter< std::vector<std::string> >("transferModes");
       for (auto & mode: modes) modesVal.append(mode);
       (*transferSystemJson_)["transferModes"]=modesVal;
 
       for (auto psKeyItr =tsPset.psetTable().begin();psKeyItr!=tsPset.psetTable().end(); ++ psKeyItr) {
         if (psKeyItr->first!="destinations" && psKeyItr->first!="transferModes") {
           const edm::ParameterSet & streamDef = tsPset.getParameterSet(psKeyItr->first);
           Json::Value streamVal;
           for (auto & mode : modes) {
             //validation
             if (!streamDef.existsAs<std::vector<std::string>>(mode,true))
               throw cms::Exception("EvFDaqDirector") << " Missing transfer system specification for -:" << psKeyItr->first << " (transferMode " << mode << ")";
             std::vector<std::string> streamDestinations =  streamDef.getParameter<std::vector<std::string>>(mode);
 
             Json::Value sDestsValue(Json::arrayValue);
 
             if (!streamDestinations.size())
               throw cms::Exception("EvFDaqDirector") << " Missing transter system destination(s) for -: "<< psKeyItr->first << ", mode:" << mode;
 
             for (auto & sdest:streamDestinations) {
               bool sDestValid=false;
               sDestsValue.append(sdest);
               for (auto & dest: destinations) {
                 if (dest==sdest) sDestValid=true;
               }
               if (!sDestValid)
                 throw cms::Exception("EvFDaqDirector") << " Invalid transter system destination specified for -: "<< psKeyItr->first << ", mode:" << mode << ", dest:"<<sdest;
             }
             streamVal[mode]=sDestsValue;
           }
           (*transferSystemJson_)[psKeyItr->first] = streamVal;
         }
       }
     }
     else {
       if (requireTSPSet_)
         throw cms::Exception("EvFDaqDirector") << "transferSystem PSet not found";
     }
   }

void evf::EvFDaqDirector::createProcessingNotificationMaybe ( ) const

Definition at line 1043 of file EvFDaqDirector.cc.

References run_dir_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::checkNextEvent().

                                                                {
     std::string proc_flag = run_dir_ + "/processing";
     int proc_flag_fd = open(proc_flag.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
     close(proc_flag_fd);
   }

void evf::EvFDaqDirector::createRunOpendirMaybe ( )

Definition at line 884 of file EvFDaqDirector.cc.

References getRunOpenDirPath(), LogDebug, and fed_dqm_sourceclient-live_cfg::path.

Referenced by initRun().

                                              {
     // 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);
     }
   }

bool evf::EvFDaqDirector::emptyLumisectionMode ( ) const

inline

Definition at line 131 of file EvFDaqDirector.h.

References emptyLumisectionMode_.

Referenced by FedRawDataInputSource::checkNextEvent().

131 {return emptyLumisectionMode_;}

evf::EvFDaqDirector::emptyLumisectionMode_

bool emptyLumisectionMode_

Definition: EvFDaqDirector.h:157

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

private

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

private

void evf::EvFDaqDirector::fillDescriptions ( edm::ConfigurationDescriptions & descriptions )

static

Definition at line 286 of file EvFDaqDirector.cc.

References edm::ConfigurationDescriptions::add(), edm::ParameterSetDescription::addUntracked(), edm::ParameterSetDescription::setAllowAnything(), edm::ParameterSetDescription::setComment(), and AlCaHLTBitMon_QueryRunRegistry::string.

   {
     edm::ParameterSetDescription desc;
     desc.setComment("Service used for file locking arbitration and for propagating information between other EvF components");
     desc.addUntracked<std::string> ("baseDir", ".")->setComment("Local base directory for run output");
     desc.addUntracked<std::string> ("buBaseDir", ".")->setComment("BU base ramdisk directory ");
     desc.addUntracked<unsigned int> ("runNumber",0)->setComment("Run Number in ramdisk to open");
     desc.addUntracked<bool>("outputAdler32Recheck",false)->setComment("Check Adler32 of per-process output files while micro-merging");
     desc.addUntracked<bool>("requireTransfersPSet",false)->setComment("Require complete transferSystem PSet in the process configuration");
     desc.addUntracked<std::string>("selectedTransferMode","")->setComment("Selected transfer mode (choice in Lvl0 propagated as Python parameter");
     desc.addUntracked<unsigned int>("fuLockPollInterval",2000)->setComment("Lock polling interval in microseconds for the input directory file lock");
     desc.addUntracked<bool>("emptyLumisectionMode",true)->setComment("Enables writing stream output metadata even when no events are processed in a lumisection");
     desc.addUntracked<bool>("microMergeDisabled",true)->setComment("Disabled micro-merging by the Output Module, so it is later done by hltd service");
     desc.addUntracked<std::string>("mergingPset","")->setComment("Name of merging PSet to look for merging type definitions for streams");
     desc.setAllowAnything();
     descriptions.add("EvFDaqDirector", desc);
   }

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

inline

Definition at line 76 of file EvFDaqDirector.h.

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

76 { return dirManager_.findRunDir(run_);}

evf::EvFDaqDirector::dirManager_

DirManager dirManager_

Definition: EvFDaqDirector.h:181

evf::EvFDaqDirector::run_

unsigned int run_

Definition: EvFDaqDirector.h:151

evf::DirManager::findRunDir

std::string findRunDir(unsigned int)

Definition: DirManager.cc:37

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

Definition at line 461 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::createBoLSFile().

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

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

Definition at line 397 of file EvFDaqDirector.cc.

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

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

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

Definition at line 453 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 457 of file EvFDaqDirector.cc.

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

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

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

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

Definition at line 465 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 470 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 425 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 380 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::getMergedDatChecksumFilePath	(	const unsigned int	ls,
		std::string const &	stream
	)		const

Definition at line 417 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 413 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 437 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 449 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 665 of file EvFDaqDirector.cc.

References bu_run_dir_, data, def, jsoncollector::DataPoint::deserialize(), reco::dp, 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 401 of file EvFDaqDirector.cc.

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

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

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

Definition at line 421 of file EvFDaqDirector.cc.

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

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

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

Definition at line 393 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 405 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 429 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 389 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 441 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 409 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 433 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 385 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 445 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 107 of file EvFDaqDirector.h.

References run_.

107 { return run_; }

evf::EvFDaqDirector::run_

unsigned int run_

Definition: EvFDaqDirector.h:151

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

inline

Definition at line 100 of file EvFDaqDirector.h.

References run_dir_.

Referenced by createRunOpendirMaybe(), and initRun().

100 {return run_dir_ +"/open";}

evf::EvFDaqDirector::run_dir_

std::string run_dir_

Definition: EvFDaqDirector.h:163

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

Definition at line 968 of file EvFDaqDirector.cc.

References Json::Value::begin(), Json::Value::end(), Exception, visualization-live-secondInstance_cfg::msg, requireTSPSet_, runTheMatrix::ret, selectedTransferMode_, AlCaHLTBitMon_QueryRunRegistry::string, and transferSystemJson_.

   {
     std::string streamRequestName;
     if (transferSystemJson_->isMember(stream.c_str()))
       streamRequestName = stream;
     else {
       std::stringstream msg;
       msg << "Transfer system mode definitions missing for -: " << stream;
       if (requireTSPSet_)
         throw cms::Exception("EvFDaqDirector") << msg.str();
       else {
         edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
         return std::string("Failsafe");
       }
     }
     //return empty if strict check parameter is not on
     if (!requireTSPSet_ && (selectedTransferMode_=="" || selectedTransferMode_=="null")) {
       edm::LogWarning("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter."
                                         << "Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
       return std::string("Failsafe");
     }
     if (requireTSPSet_ && (selectedTransferMode_=="" || selectedTransferMode_=="null")) {
       throw cms::Exception("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter.";
     }
     //check if stream has properly listed transfer stream
     if  (!transferSystemJson_->get(streamRequestName, "").isMember(selectedTransferMode_.c_str()))
     {
          std::stringstream msg;
          msg << "Selected transfer mode " << selectedTransferMode_ << " is not specified for stream " << streamRequestName;
          if (requireTSPSet_)
            throw cms::Exception("EvFDaqDirector") << msg.str();
          else
            edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)"; 
            return std::string("Failsafe");
     }
     Json::Value destsVec = transferSystemJson_->get(streamRequestName, "").get(selectedTransferMode_,"");
 
     //flatten string json::Array into CSV std::string
     std::string ret;
     for (Json::Value::iterator it = destsVec.begin(); it!=destsVec.end(); it++)
     {
       if (ret!="") ret +=",";
       ret+=(*it).asString();
     }
     return ret;
   }

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

inline

Definition at line 114 of file EvFDaqDirector.h.

References streamFileTracker_.

Referenced by FedRawDataInputSource::getNextEvent().

114 {return &streamFileTracker_;}

evf::EvFDaqDirector::streamFileTracker_

std::vector< int > streamFileTracker_

Definition: EvFDaqDirector.h:199

std::string evf::EvFDaqDirector::getStreamMergeType	(	std::string const &	stream,
		MergeType	defaultType
	)

Definition at line 1030 of file EvFDaqDirector.cc.

References assert(), mergeTypeMap_, MergeTypeNames_, and AlCaHLTBitMon_QueryRunRegistry::string.

   {
     auto mergeTypeItr = mergeTypeMap_.find(stream.c_str());
     if (mergeTypeItr == mergeTypeMap_.end()) {
            edm::LogInfo("EvFDaqDirector") << " No merging type specified for stream " << stream << ". Using default value";
            assert(defaultType<MergeTypeNames_.size());
            std::string defaultName = MergeTypeNames_[defaultType];
            mergeTypeMap_[stream] =  defaultName;
            return defaultName;
     }
     return mergeTypeItr->second;
   }

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

private

void evf::EvFDaqDirector::initRun ( void )

Definition at line 130 of file EvFDaqDirector.cc.

References base_dir_, bu_base_dir_, bu_run_dir_, bu_run_open_dir_, bu_w_lock_stream, bu_writelock_fd_, eostools::chmod(), createRunOpendirMaybe(), directorBu_, Exception, fu_readwritelock_fd_, fu_rw_lock_stream, fulocal_rwlock_fd2_, fulocal_rwlock_fd_, getRunOpenDirPath(), hltSourceDirectory_, init_lock_, eostools::mkdir(), NULL, openFULockfileStream(), run_, run_dir_, run_string_, contentValuesCheck::ss, stopFilePath_, stopFilePathPid_, AlCaHLTBitMon_QueryRunRegistry::string, and tryInitializeFuLockFile().

Referenced by preallocate().

   {
     std::stringstream ss;
     ss << "run" << std::setfill('0') << std::setw(6) << run_;
     run_string_ = ss.str();
     run_dir_ = base_dir_+"/"+run_string_;
 
     // 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_);
 
         if (hltSourceDirectory_.size())
     {
       struct stat buf;
       if (stat(hltSourceDirectory_.c_str(),&buf)==0) {
         std::string hltdir=bu_run_dir_+"/hlt";
         std::string tmphltdir=bu_run_open_dir_+"/hlt";
         retval = mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
         if (retval != 0 && errno != EEXIST)
           throw cms::Exception("DaqDirector")
             << " Error creating bu run dir -: " << hltdir
             << " mkdir error:" << strerror(errno) << "\n";
 
             boost::filesystem::copy_file(hltSourceDirectory_+"/HltConfig.py",tmphltdir+"/HltConfig.py");
 
             boost::filesystem::copy_file(hltSourceDirectory_+"/fffParameters.jsn",tmphltdir+"/fffParameters.jsn");
 
             boost::filesystem::rename(tmphltdir,hltdir);
       }
           else
         throw cms::Exception("DaqDirector") << " Error looking for HLT configuration -: " << hltSourceDirectory_;
     }
         //else{}//no configuration specified
       }
     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";
     std::stringstream sstp;
     sstp << stopFilePath_ << "_pid" << getpid();
     stopFilePathPid_ = sstp.str();
   }

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

private

bool evf::EvFDaqDirector::isSingleStreamThread ( )

inline

Definition at line 115 of file EvFDaqDirector.h.

References nStreams_, and nThreads_.

Referenced by FedRawDataInputSource::getNextEvent().

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

evf::EvFDaqDirector::nThreads_

unsigned int nThreads_

Definition: EvFDaqDirector.h:208

evf::EvFDaqDirector::nStreams_

unsigned int nStreams_

Definition: EvFDaqDirector.h:207

void evf::EvFDaqDirector::lockFULocal ( )

Definition at line 862 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_SH);
   }

void evf::EvFDaqDirector::lockFULocal2 ( )

Definition at line 873 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 854 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 834 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

bool evf::EvFDaqDirector::microMergeDisabled ( ) const

inline

Definition at line 132 of file EvFDaqDirector.h.

References microMergeDisabled_.

132 {return microMergeDisabled_;}

evf::EvFDaqDirector::microMergeDisabled_

bool microMergeDisabled_

Definition: EvFDaqDirector.h:158

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

private

Definition at line 815 of file EvFDaqDirector.cc.

References eostools::chmod(), fu_readwritelock_fd_, fu_rw_lock_stream, and LogDebug.

Referenced by initRun().

                                                                               {
     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 101 of file EvFDaqDirector.h.

References outputAdler32Recheck_.

101 {return outputAdler32Recheck_;}

evf::EvFDaqDirector::outputAdler32Recheck_

bool outputAdler32Recheck_

Definition: EvFDaqDirector.h:152

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

private

void evf::EvFDaqDirector::overrideRunNumber ( unsigned int run )

inline

Definition at line 71 of file EvFDaqDirector.h.

References DTTTrigCorrFirst::run, and run_.

71 {run_=run;}

evf::EvFDaqDirector::run_

unsigned int run_

Definition: EvFDaqDirector.h:151

DTTTrigCorrFirst.run

run

Definition: DTTTrigCorrFirst.py:63

void evf::EvFDaqDirector::postEndRun ( edm::GlobalContext const & globalContext )

Definition at line 322 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 275 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

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

void evf::EvFDaqDirector::preBeginJob	(	edm::PathsAndConsumesOfModulesBase const &	,
		edm::ProcessContext const &	pc
	)

Definition at line 304 of file EvFDaqDirector.cc.

References checkMergeTypePSet(), and checkTransferSystemPSet().

Referenced by EvFDaqDirector().

                                                                 {
     checkTransferSystemPSet(pc);
     checkMergeTypePSet(pc);
   }

void evf::EvFDaqDirector::preBeginRun ( edm::GlobalContext const & globalContext )

Definition at line 310 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 331 of file EvFDaqDirector.cc.

References cppFunctionSkipper::exception, fileDeleteLockPtr_, filesToDeletePtr_, LogDebug, eostools::ls(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), fed_dqm_sourceclient-live_cfg::path, and MatrixUtil::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 (const boost::filesystem::filesystem_error&) {/*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 (std::exception&) {/*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 375 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 895 of file EvFDaqDirector.cc.

References Json::Value::asInt(), mergeVDriftHistosByStation::file, Json::Value::get(), Json::Reader::parse(), matplotRender::reader, and runTheMatrix::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 481 of file EvFDaqDirector.cc.

References getRawFilePath().

Referenced by postEndRun().

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

void evf::EvFDaqDirector::removeFile ( std::string filename )

Definition at line 474 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 123 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 112 of file EvFDaqDirector.h.

References fms_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

112 {fms_=fms;}

evf::EvFDaqDirector::fms_

evf::FastMonitoringService * fms_

Definition: EvFDaqDirector.h:198

void evf::EvFDaqDirector::tryInitializeFuLockFile ( )

Definition at line 804 of file EvFDaqDirector.cc.

References fu_rw_lock_stream, and getHLTprescales::readIndex().

Referenced by initRun().

                                                {
     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;
       fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
     }
   }

void evf::EvFDaqDirector::unlockAndCloseMergeStream ( )

Definition at line 848 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 867 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 878 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

Referenced by FedRawDataInputSource::checkNextEvent(), 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 858 of file EvFDaqDirector.cc.

References init_lock_.

                                       {
     pthread_mutex_unlock(&init_lock_);
   }

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

inline

Definition at line 113 of file EvFDaqDirector.h.

References currentFileIndex_.

Referenced by FedRawDataInputSource::getNextEvent().

113 {currentFileIndex_=fileIndex;}

evf::EvFDaqDirector::currentFileIndex_

int currentFileIndex_

Definition: EvFDaqDirector.h:200

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

Definition at line 485 of file EvFDaqDirector.cc.

References bu_run_dir_, bumpFile(), trackerTree::check(), Exception, fu_readwritelock_fd_, fu_rw_flk, fu_rw_fulk, fu_rw_lock_stream, fuLockPollInterval_, getEoLSFilePathOnFU(), getEoRFilePath(), lockFULocal(), LogDebug, eostools::ls(), newFile, noFile, getHLTprescales::readIndex(), readLastLSEntry(), runEnded, stop_ls_override_, stopFilePath_, and stopFilePathPid_.

Referenced by FedRawDataInputSource::readSupervisor().

                                                                                                                                       {
     EvFDaqDirector::FileStatus fileStatus = noFile;
 
     int retval = -1;
     int lock_attempts = 0;
 
     struct stat buf;
     int stopFileLS = -1;
     int stopFileCheck = stat(stopFilePath_.c_str(),&buf);
     int stopFilePidCheck = stat(stopFilePathPid_.c_str(),&buf);
     if (stopFileCheck==0 || stopFilePidCheck==0) {
         if (stopFileCheck==0)
           stopFileLS = readLastLSEntry(stopFilePath_);
         else
           stopFileLS = 1;//stop without drain if only pid is stopped
         if (!stop_ls_override_) {
           //if lumisection is higher than in stop file, should quit at next from current
           if (stopFileLS>=0 && (int)ls>=stopFileLS) stopFileLS = stop_ls_override_ = ls;
         }
         else stopFileLS = stop_ls_override_;
         edm::LogWarning("EvFDaqDirector") << "Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
         //return runEnded;
     }
     else //if file was removed before reaching stop condition, reset this
       stop_ls_override_ = 0;
 
     timeval ts_lockbegin;
     gettimeofday(&ts_lockbegin,0);
 
     while (retval==-1) {
       retval = fcntl(fu_readwritelock_fd_, F_SETLK, &fu_rw_flk);
       if (retval==-1) usleep(fuLockPollInterval_);
       else continue;
 
       lock_attempts+=fuLockPollInterval_;
       if (lock_attempts>5000000 ||  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(getEoLSFilePathOnFU(ls).c_str(),&buf)==0) {
           edm::LogWarning("EvFDaqDirector") << "Detected local EoLS for lumisection "<< ls ; 
           ls++;
           return noFile;
         }
 
         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;
       }
     }
 
     timeval ts_lockend;
     gettimeofday(&ts_lockend,0);
     long deltat = (ts_lockend.tv_usec-ts_lockbegin.tv_usec) + (ts_lockend.tv_sec-ts_lockbegin.tv_sec)*1000000;
     if (deltat>0.) lockWaitTime=deltat;
 
     
  
     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;
       int check = 0;
       // rewind the stream
       check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
       // if rewinded ok
       if (check == 0) {
     // read its' values
     fscanf(fu_rw_lock_stream, "%u %u", &readLs, &readIndex);
     edm::LogInfo("EvFDaqDirector") << "Read fu.lock file file -: " << readLs << ":" << readIndex;
 
         unsigned int currentLs = readLs;
         bool bumpedOk = false;
         //if next lumisection in a lock file is not +1 wrt. source, cycle through the next empty one, unless initial lumi not yet set
         //no lock file write in this case
         if (ls && ls+1 < currentLs) ls++;
         else {
       // try to bump (look for new index or EoLS file)
       bumpedOk = bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
           //avoid 2 lumisections jump
           if (ls && readLs>currentLs && currentLs > ls) {
             ls++;
             readLs=currentLs=ls;
             readIndex=0;
             bumpedOk=false;
             //no write to lock file
           }
           else {
             if (ls==0 && readLs>currentLs) {
               //make sure to intialize always with LS found in the lock file, with possibility of grabbing index file immediately
               //in this case there is no new file in the same LS
               readLs=currentLs;
               readIndex=0;
               bumpedOk=false;
               //no write to lock file
             }
             //update return LS value
         ls = readLs;
           }
         }
     if (bumpedOk) {
       // there is a new index file to grab, lock file needs to be updated
       check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
       if (check == 0) {
         ftruncate(fu_readwritelock_fd_, 0);
         // write next index in the file, which is the file the next process should take
         fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex + 1);
         fflush(fu_rw_lock_stream);
         fsync(fu_readwritelock_fd_);
         fileStatus = newFile;
         LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":" << readIndex + 1;
       }
           else {
             throw cms::Exception("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error " << strerror(errno);
           }
     }
         else if (currentLs < readLs) {
           //there is no new file in next LS (yet), but lock file can be updated to the next LS
       check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
       if (check == 0) {
         ftruncate(fu_readwritelock_fd_, 0);
         // in this case LS was bumped, but no new file. Thus readIndex is 0 (set by bumpFile)
         fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
         fflush(fu_rw_lock_stream);
         fsync(fu_readwritelock_fd_);
         LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":" << readIndex;
       }
           else {
             throw cms::Exception("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 148 of file EvFDaqDirector.h.

Referenced by initRun().

std::string evf::EvFDaqDirector::bu_base_dir_

private

Definition at line 149 of file EvFDaqDirector.h.

Referenced by initRun().

struct flock evf::EvFDaqDirector::bu_r_flk

private

Definition at line 186 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_r_fulk

private

Definition at line 188 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_r_lock_stream

private

Definition at line 175 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_readlock_fd_

private

Definition at line 167 of file EvFDaqDirector.h.

Referenced by postEndRun().

std::string evf::EvFDaqDirector::bu_run_dir_

private

Definition at line 164 of file EvFDaqDirector.h.

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

std::string evf::EvFDaqDirector::bu_run_open_dir_

private

Definition at line 165 of file EvFDaqDirector.h.

Referenced by buBaseRunOpenDir(), and initRun().

FILE* evf::EvFDaqDirector::bu_t_monitor_stream

private

Definition at line 178 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_flk

private

Definition at line 185 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_fulk

private

Definition at line 187 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_w_lock_stream

private

Definition at line 174 of file EvFDaqDirector.h.

Referenced by initRun().

FILE* evf::EvFDaqDirector::bu_w_monitor_stream

private

Definition at line 177 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_writelock_fd_

private

Definition at line 168 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

int evf::EvFDaqDirector::currentFileIndex_ = -1

private

Definition at line 200 of file EvFDaqDirector.h.

Referenced by preSourceEvent(), and updateFileIndex().

int evf::EvFDaqDirector::data_readwrite_fd_

private

Definition at line 170 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream(), and unlockAndCloseMergeStream().

struct flock evf::EvFDaqDirector::data_rw_flk

private

Definition at line 191 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream().

struct flock evf::EvFDaqDirector::data_rw_fulk

private

Definition at line 192 of file EvFDaqDirector.h.

Referenced by unlockAndCloseMergeStream().

FILE* evf::EvFDaqDirector::data_rw_stream

private

Definition at line 179 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream(), and unlockAndCloseMergeStream().

bool evf::EvFDaqDirector::directorBu_

private

Definition at line 150 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

DirManager evf::EvFDaqDirector::dirManager_

private

Definition at line 181 of file EvFDaqDirector.h.

Referenced by findCurrentRunDir(), and preBeginRun().

bool evf::EvFDaqDirector::emptyLumisectionMode_

private

Definition at line 157 of file EvFDaqDirector.h.

Referenced by emptyLumisectionMode(), and EvFDaqDirector().

unsigned int evf::EvFDaqDirector::eolsNFilesIndex_ = 1

private

Definition at line 211 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

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

private

Definition at line 202 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

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

private

Definition at line 203 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

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

private

Definition at line 198 of file EvFDaqDirector.h.

Referenced by bumpFile(), and setFMS().

int evf::EvFDaqDirector::fu_readwritelock_fd_

private

Definition at line 169 of file EvFDaqDirector.h.

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

struct flock evf::EvFDaqDirector::fu_rw_flk

private

Definition at line 189 of file EvFDaqDirector.h.

Referenced by updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_fulk

private

Definition at line 190 of file EvFDaqDirector.h.

Referenced by updateFuLock().

FILE* evf::EvFDaqDirector::fu_rw_lock_stream

private

Definition at line 176 of file EvFDaqDirector.h.

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

int evf::EvFDaqDirector::fulocal_rwlock_fd2_

private

Definition at line 172 of file EvFDaqDirector.h.

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

int evf::EvFDaqDirector::fulocal_rwlock_fd_

private

Definition at line 171 of file EvFDaqDirector.h.

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

unsigned int evf::EvFDaqDirector::fuLockPollInterval_

private

Definition at line 156 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateFuLock().

std::string evf::EvFDaqDirector::hltSourceDirectory_

private

Definition at line 155 of file EvFDaqDirector.h.

Referenced by initRun().

std::string evf::EvFDaqDirector::hostname_

private

Definition at line 161 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 205 of file EvFDaqDirector.h.

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

std::map<std::string,std::string> evf::EvFDaqDirector::mergeTypeMap_

private

Definition at line 217 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet(), and getStreamMergeType().

const std::vector< std::string > evf::EvFDaqDirector::MergeTypeNames_ = {"","DAT","PB","JSNDATA"}

staticprivate

Definition at line 220 of file EvFDaqDirector.h.

Referenced by getStreamMergeType().

std::string evf::EvFDaqDirector::mergeTypePset_

private

Definition at line 159 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet().

bool evf::EvFDaqDirector::microMergeDisabled_

private

Definition at line 158 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and microMergeDisabled().

unsigned int evf::EvFDaqDirector::nStreams_ =0

private

Definition at line 207 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

unsigned int evf::EvFDaqDirector::nThreads_ =0

private

Definition at line 208 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

bool evf::EvFDaqDirector::outputAdler32Recheck_

private

Definition at line 152 of file EvFDaqDirector.h.

Referenced by outputAdler32Recheck().

unsigned long evf::EvFDaqDirector::previousFileSize_

private

Definition at line 183 of file EvFDaqDirector.h.

Referenced by bumpFile().

bool evf::EvFDaqDirector::readEolsDefinition_ = true

private

Definition at line 210 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

bool evf::EvFDaqDirector::requireTSPSet_

private

Definition at line 153 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().

unsigned int evf::EvFDaqDirector::run_

private

std::string evf::EvFDaqDirector::run_dir_

private

std::string evf::EvFDaqDirector::run_string_

private

Definition at line 162 of file EvFDaqDirector.h.

Referenced by initRun().

std::string evf::EvFDaqDirector::selectedTransferMode_

private

Definition at line 154 of file EvFDaqDirector.h.

Referenced by getStreamDestinations().

unsigned int evf::EvFDaqDirector::stop_ls_override_ = 0

private

Definition at line 214 of file EvFDaqDirector.h.

Referenced by updateFuLock().

std::string evf::EvFDaqDirector::stopFilePath_

private

Definition at line 212 of file EvFDaqDirector.h.

Referenced by initRun(), and updateFuLock().

std::string evf::EvFDaqDirector::stopFilePathPid_

private

Definition at line 213 of file EvFDaqDirector.h.

Referenced by initRun(), and updateFuLock().

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

private

Definition at line 199 of file EvFDaqDirector.h.

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

std::shared_ptr<Json::Value> evf::EvFDaqDirector::transferSystemJson_

private

Definition at line 216 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().

Public Types

Public Member Functions

Static Public Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Member Enumeration Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation