#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)

EvFDaqDirector::FileStatus	contactFileBroker (unsigned int &serverHttpStatus, bool &serverState, uint32_t &serverLS, uint32_t &closedServerLS, std::string &nextFileJson, std::string &nextFileRaw, bool &rawHeader, int maxLS)

void	createBoLSFile (const uint32_t lumiSection, bool checkIfExists) const

void	createLumiSectionFiles (const uint32_t lumiSection, const uint32_t currentLumiSection, bool doCreateBoLS, bool doCreateEoLS)

void	createProcessingNotificationMaybe () const

void	createRunOpendirMaybe ()

	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	getFFFParamsFilePathOnBU () const

std::string	getInitFilePath (std::string const &stream) const

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

unsigned int	getLumisectionToStart () 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

FileStatus	getNextFromFileBroker (const unsigned int currentLumiSection, unsigned int &ls, std::string &nextFile, int &rawFd, uint16_t &rawHeaderSize, int32_t &serverEventsInNewFile_, int64_t &fileSize, uint64_t &thisLockWaitTimeUs)

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

unsigned int	getStartLumisectionFromEnv () const

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

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

int	grabNextJsonFile (std::string const &jsonSourcePath, std::string const &rawSourcePath, int64_t &fileSizeFromJson, bool &fileFound)

int	grabNextJsonFileAndUnlock (std::filesystem::path const &jsonSourcePath)

int	grabNextJsonFromRaw (std::string const &rawSourcePath, int &rawFd, uint16_t &rawHeaderSize, int64_t &fileSizeFromHeader, bool &fileFound, uint32_t serverLS, bool closeFile)

void	initRun ()

bool	inputThrottled ()

bool	isSingleStreamThread ()

void	lockFULocal ()

void	lockFULocal2 ()

void	lockInitLock ()

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)

bool	rawFileHasHeader (std::string const &rawSourcePath, uint16_t &rawHeaderSize)

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, std::unique_ptr< InputFile >>> filesToDelete)

void	setFMS (evf::FastMonitoringService *fms)

void	tryInitializeFuLockFile ()

void	unlockFULocal ()

void	unlockFULocal2 ()

void	unlockInitLock ()

FileStatus	updateFuLock (unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint16_t &rawHeaderSize, uint64_t &lockWaitTime, bool &setExceptionState)

bool	useFileBroker () const

	~EvFDaqDirector ()

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

static struct flock	make_flock (short type, short whence, off_t start, off_t len, pid_t pid)

static int	parseFRDFileHeader (std::string const &rawSourcePath, int &rawFd, uint16_t &rawHeaderSize, uint32_t &lsFromHeader, int32_t &eventsFromHeader, int64_t &fileSizeFromHeader, bool requireHeader, bool retry, bool closeFile)

Private Member Functions
bool	bumpFile (unsigned int &ls, unsigned int &index, std::string &nextFile, uint32_t &fsize, uint16_t &rawHeaderSize, int maxLS, bool &setExceptionState)

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 (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_

bool	directorBU_

DirManager	dirManager_

jsoncollector::DataPointDefinition *	dpd_

std::unique_ptr< boost::asio::ip::tcp::resolver::iterator >	endpoint_iterator_

unsigned int	eolsNFilesIndex_ = 1

std::string	fileBrokerHost_

bool	fileBrokerHostFromCfg_

bool	fileBrokerKeepAlive_

std::string	fileBrokerPort_

bool	fileBrokerUseLocalLock_

std::mutex *	fileDeleteLockPtr_ = nullptr

std::list< std::pair< int, std::unique_ptr< 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	fulockfile_

unsigned int	fuLockPollInterval_

std::string	hltSourceDirectory_

std::string	hostname_

pthread_mutex_t	init_lock_ = PTHREAD_MUTEX_INITIALIZER

std::string	input_throttled_file_

boost::asio::io_service	io_service_

tbb::concurrent_hash_map< std::string, std::string >	mergeTypeMap_

std::string	mergeTypePset_

unsigned int	nStreams_ = 0

unsigned int	nThreads_ = 0

bool	outputAdler32Recheck_

std::string	pid_

unsigned long	previousFileSize_

std::unique_ptr< boost::asio::ip::tcp::resolver::query >	query_

bool	readEolsDefinition_ = true

bool	requireTSPSet_

std::unique_ptr< boost::asio::ip::tcp::resolver >	resolver_

unsigned int	run_

std::string	run_dir_

std::string	run_nstring_

std::string	run_string_

std::string	selectedTransferMode_

std::unique_ptr< boost::asio::ip::tcp::socket >	socket_

unsigned int	startFromLS_ = 1

unsigned int	stop_ls_override_ = 0

std::string	stopFilePath_

std::string	stopFilePathPid_

std::shared_ptr< Json::Value >	transferSystemJson_

bool	useFileBroker_

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

Detailed Description

Definition at line 62 of file EvFDaqDirector.h.

Member Enumeration Documentation

◆ FileStatus

enum evf::EvFDaqDirector::FileStatus

Enumerator
noFile
sameFile
newFile
newLumi
runEnded
runAbort

Definition at line 64 of file EvFDaqDirector.h.

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

evf::EvFDaqDirector::runAbort

Definition: EvFDaqDirector.h:64

evf::EvFDaqDirector::sameFile

Definition: EvFDaqDirector.h:64

evf::EvFDaqDirector::newLumi

Definition: EvFDaqDirector.h:64

evf::EvFDaqDirector::noFile

Definition: EvFDaqDirector.h:64

evf::EvFDaqDirector::newFile

Definition: EvFDaqDirector.h:64

evf::EvFDaqDirector::runEnded

Definition: EvFDaqDirector.h:64

Constructor & Destructor Documentation

◆ EvFDaqDirector()

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

explicit

Definition at line 38 of file EvFDaqDirector.cc.

References visDQMUpload::buf, endpoint_iterator_, cppFunctionSkipper::exception, fileBrokerHost_, fileBrokerHostFromCfg_, fileBrokerPort_, fileBrokerUseLocalLock_, fuLockPollInterval_, hostname_, recoMuon::in, io_service_, postEndRun(), preallocate(), preBeginJob(), preBeginRun(), preGlobalEndLumi(), query_, resolver_, socket_, startFromLS_, edm_modernize_messagelogger::stat, AlCaHLTBitMon_QueryRunRegistry::string, useFileBroker_, edm::ActivityRegistry::watchPostGlobalEndRun(), edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreBeginJob(), edm::ActivityRegistry::watchPreGlobalBeginRun(), and edm::ActivityRegistry::watchPreGlobalEndLumi().

       : base_dir_(pset.getUntrackedParameter<std::string>("baseDir")),
         bu_base_dir_(pset.getUntrackedParameter<std::string>("buBaseDir")),
         run_(pset.getUntrackedParameter<unsigned int>("runNumber")),
         useFileBroker_(pset.getUntrackedParameter<bool>("useFileBroker")),
         fileBrokerHostFromCfg_(pset.getUntrackedParameter<bool>("fileBrokerHostFromCfg", true)),
         fileBrokerHost_(pset.getUntrackedParameter<std::string>("fileBrokerHost", "InValid")),
         fileBrokerPort_(pset.getUntrackedParameter<std::string>("fileBrokerPort", "8080")),
         fileBrokerKeepAlive_(pset.getUntrackedParameter<bool>("fileBrokerKeepAlive", true)),
         fileBrokerUseLocalLock_(pset.getUntrackedParameter<bool>("fileBrokerUseLocalLock", true)),
         fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>("fuLockPollInterval", 2000)),
         outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck", false)),
         requireTSPSet_(pset.getUntrackedParameter<bool>("requireTransfersPSet", false)),
         selectedTransferMode_(pset.getUntrackedParameter<std::string>("selectedTransferMode", "")),
         mergeTypePset_(pset.getUntrackedParameter<std::string>("mergingPset", "")),
         directorBU_(pset.getUntrackedParameter<bool>("directorIsBU", false)),
         hltSourceDirectory_(pset.getUntrackedParameter<std::string>("hltSourceDirectory", "")),
         hostname_(""),
         bu_readlock_fd_(-1),
         bu_writelock_fd_(-1),
         fu_readwritelock_fd_(-1),
         fulocal_rwlock_fd_(-1),
         fulocal_rwlock_fd2_(-1),
         bu_w_lock_stream(nullptr),
         bu_r_lock_stream(nullptr),
         fu_rw_lock_stream(nullptr),
         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())) {
     reg.watchPreallocate(this, &EvFDaqDirector::preallocate);
     reg.watchPreBeginJob(this, &EvFDaqDirector::preBeginJob);
     reg.watchPreGlobalBeginRun(this, &EvFDaqDirector::preBeginRun);
     reg.watchPostGlobalEndRun(this, &EvFDaqDirector::postEndRun);
     reg.watchPreGlobalEndLumi(this, &EvFDaqDirector::preGlobalEndLumi);
 
     //save hostname for later
     char hostname[33];
     gethostname(hostname, 32);
     hostname_ = hostname;
 
     char* fuLockPollIntervalPtr = std::getenv("FFF_LOCKPOLLINTERVAL");
     if (fuLockPollIntervalPtr) {
       try {
         fuLockPollInterval_ = std::stoul(std::string(fuLockPollIntervalPtr));
         edm::LogInfo("EvFDaqDirector") << "Setting fu lock poll interval by environment string: " << fuLockPollInterval_
                                        << " us";
       } catch (const std::exception&) {
         edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fuLockPollIntervalPtr);
       }
     }
 
     //override file service parameter if specified by environment
     char* fileBrokerParamPtr = std::getenv("FFF_USEFILEBROKER");
     if (fileBrokerParamPtr) {
       try {
         useFileBroker_ = (std::stoul(std::string(fileBrokerParamPtr))) > 0;
         edm::LogInfo("EvFDaqDirector") << "Setting useFileBroker parameter by environment string: " << useFileBroker_;
       } catch (const std::exception&) {
         edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fileBrokerParamPtr);
       }
     }
     if (useFileBroker_) {
       if (fileBrokerHostFromCfg_) {
         //find BU data address from hltd configuration
         fileBrokerHost_ = std::string();
         struct stat buf;
         if (stat("/etc/appliance/bus.config", &buf) == 0) {
           std::ifstream busconfig("/etc/appliance/bus.config", std::ifstream::in);
           std::getline(busconfig, fileBrokerHost_);
         }
         if (fileBrokerHost_.empty())
           throw cms::Exception("EvFDaqDirector") << "No file service or BU data address information";
       } else if (fileBrokerHost_.empty() || fileBrokerHost_ == "InValid")
         throw cms::Exception("EvFDaqDirector")
             << "fileBrokerHostFromCfg must be set to true if fileBrokerHost parameter is not valid or empty";
 
       resolver_ = std::make_unique<boost::asio::ip::tcp::resolver>(io_service_);
       query_ = std::make_unique<boost::asio::ip::tcp::resolver::query>(fileBrokerHost_, fileBrokerPort_);
       endpoint_iterator_ = std::make_unique<boost::asio::ip::tcp::resolver::iterator>(resolver_->resolve(*query_));
       socket_ = std::make_unique<boost::asio::ip::tcp::socket>(io_service_);
     }
 
     char* startFromLSPtr = std::getenv("FFF_START_LUMISECTION");
     if (startFromLSPtr) {
       try {
         startFromLS_ = std::stoul(std::string(startFromLSPtr));
         edm::LogInfo("EvFDaqDirector") << "Setting start from LS by environment string: " << startFromLS_;
       } catch (const std::exception&) {
         edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(startFromLSPtr);
       }
     }
 
     //override file service parameter if specified by environment
     char* fileBrokerUseLockParamPtr = std::getenv("FFF_FILEBROKERUSELOCALLOCK");
     if (fileBrokerUseLockParamPtr) {
       try {
         fileBrokerUseLocalLock_ = (std::stoul(std::string(fileBrokerUseLockParamPtr))) > 0;
         edm::LogInfo("EvFDaqDirector") << "Setting fileBrokerUseLocalLock parameter by environment string: "
                                        << fileBrokerUseLocalLock_;
       } catch (const std::exception&) {
         edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fileBrokerUseLockParamPtr);
       }
     }
   }

◆ ~EvFDaqDirector()

evf::EvFDaqDirector::~EvFDaqDirector ( )

Definition at line 301 of file EvFDaqDirector.cc.

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

                                   {
     //close server connection
     if (socket_.get() && socket_->is_open()) {
       boost::system::error_code ec;
       socket_->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
       socket_->close(ec);
     }
 
     if (fulocal_rwlock_fd_ != -1) {
       unlockFULocal();
       close(fulocal_rwlock_fd_);
     }
 
     if (fulocal_rwlock_fd2_ != -1) {
       unlockFULocal2();
       close(fulocal_rwlock_fd2_);
     }
   }

Member Function Documentation

◆ baseRunDir()

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

inline

Definition at line 76 of file EvFDaqDirector.h.

References run_dir_.

Referenced by grabNextJsonFile(), grabNextJsonFileAndUnlock(), and grabNextJsonFromRaw().

76 { return run_dir_; }

evf::EvFDaqDirector::run_dir_

std::string run_dir_

Definition: EvFDaqDirector.h:229

◆ buBaseRunDir()

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

inline

Definition at line 77 of file EvFDaqDirector.h.

References bu_run_dir_.

77 { return bu_run_dir_; }

evf::EvFDaqDirector::bu_run_dir_

std::string bu_run_dir_

Definition: EvFDaqDirector.h:230

◆ buBaseRunOpenDir()

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

inline

Definition at line 78 of file EvFDaqDirector.h.

References bu_run_open_dir_.

78 { return bu_run_open_dir_; }

evf::EvFDaqDirector::bu_run_open_dir_

std::string bu_run_open_dir_

Definition: EvFDaqDirector.h:231

◆ bumpFile()

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

private

Definition at line 779 of file EvFDaqDirector.cc.

References evf::FastMonitoringService::accumulateFileSize(), visDQMUpload::buf, fms_, getEoLSFilePathOnBU(), getInputJsonFilePath(), getNFilesFromEoLS(), getRawFilePath(), eostools::ls(), previousFileSize_, rawFileHasHeader(), contentValuesCheck::ss, edm_modernize_messagelogger::stat, 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;
     }
     nextFile = "";
 
     //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
     std::string nextFileJson = getInputJsonFilePath(ls, index);
     if (stat(nextFileJson.c_str(), &buf) == 0) {
       fsize = previousFileSize_ = buf.st_size;
       nextFile = nextFileJson;
       return true;
     }
     // 2. No file -> lumi ended? (and how many?)
     else {
       // 3. No file -> check for standalone raw file
       std::string nextFileRaw = getRawFilePath(ls, index);
       if (stat(nextFileRaw.c_str(), &buf) == 0 && rawFileHasHeader(nextFileRaw, rawHeaderSize)) {
         fsize = previousFileSize_ = buf.st_size;
         nextFile = nextFileRaw;
         return true;
       }
 
       std::string BUEoLSFile = getEoLSFilePathOnBU(ls);
 
       if (stat(BUEoLSFile.c_str(), &buf) == 0) {
         // recheck that no raw file appeared in the meantime
         if (stat(nextFileJson.c_str(), &buf) == 0) {
           fsize = previousFileSize_ = buf.st_size;
           nextFile = nextFileJson;
           return true;
         }
         if (stat(nextFileRaw.c_str(), &buf) == 0 && rawFileHasHeader(nextFileRaw, rawHeaderSize)) {
           fsize = previousFileSize_ = buf.st_size;
           nextFile = nextFileRaw;
           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";
             setExceptionState = true;
             return false;
           }
         }
         // this lumi ended, check for files
         ++ls;
         index = 0;
 
         //reached limit
         if (maxLS >= 0 && ls > (unsigned int)maxLS)
           return false;
 
         nextFileJson = getInputJsonFilePath(ls, 0);
         nextFileRaw = getRawFilePath(ls, 0);
         if (stat(nextFileJson.c_str(), &buf) == 0) {
           // a new file was found at new lumisection, index 0
           fsize = previousFileSize_ = buf.st_size;
           nextFile = nextFileJson;
           return true;
         }
         if (stat(nextFileRaw.c_str(), &buf) == 0 && rawFileHasHeader(nextFileRaw, rawHeaderSize)) {
           fsize = previousFileSize_ = buf.st_size;
           nextFile = nextFileRaw;
           return true;
         }
         return false;
       }
     }
     // no new file found
     return false;
   }

◆ checkMergeTypePSet()

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

Definition at line 2019 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_.empty())
       return;
     edm::ParameterSet const& topPset = edm::getParameterSet(pc.parameterSetID());
     if (topPset.existsAs<edm::ParameterSet>(mergeTypePset_, true)) {
       const edm::ParameterSet& tsPset(topPset.getParameterSet(mergeTypePset_));
       for (const std::string& pname : tsPset.getParameterNames()) {
         std::string streamType = tsPset.getParameter<std::string>(pname);
         tbb::concurrent_hash_map<std::string, std::string>::accessor ac;
         mergeTypeMap_.insert(ac, pname);
         ac->second = streamType;
         ac.release();
       }
     }
   }

◆ checkTransferSystemPSet()

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

Definition at line 1911 of file EvFDaqDirector.cc.

References Json::Value::append(), Json::arrayValue, mps_fire::dest, myMessageLogger_cff::destinations, Exception, edm::ParameterSet::existsAs(), edm::ParameterSet::getParameter(), edm::ParameterSet::getParameterSet(), edm::getParameterSet(), ALCARECOPromptCalibProdSiPixelAli0T_cff::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.empty())
               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";
     }
   }

◆ contactFileBroker()

EvFDaqDirector::FileStatus evf::EvFDaqDirector::contactFileBroker	(	unsigned int &	serverHttpStatus,
		bool &	serverState,
		uint32_t &	serverLS,
		uint32_t &	closedServerLS,
		std::string &	nextFileJson,
		std::string &	nextFileRaw,
		bool &	rawHeader,
		int	maxLS
	)

Definition at line 1457 of file EvFDaqDirector.cc.

References cms::cuda::assert(), bu_run_dir_, GlobalPosition_Frontier_DevDB_cff::connect, MillePedeFileConverter_cfg::e, endpoint_iterator_, cppFunctionSkipper::exception, fileBrokerHost_, fileBrokerKeepAlive_, RecoTauValidation_cfi::header, SiStripPI::max, newFile, noFile, castor_dqm_sourceclient_file_cfg::path, pid_, fileinputsource_cfi::read, run_nstring_, runEnded, socket_, AlCaHLTBitMon_QueryRunRegistry::string, and writeEcalDQMStatus::write.

Referenced by getNextFromFileBroker().

                                                                           {
     EvFDaqDirector::FileStatus fileStatus = noFile;
     serverError = false;
 
     boost::system::error_code ec;
     try {
       while (true) {
         //socket connect
         if (!fileBrokerKeepAlive_ || !socket_->is_open()) {
           boost::asio::connect(*socket_, *endpoint_iterator_, ec);
 
           if (ec) {
             edm::LogWarning("EvFDaqDirector") << "boost::asio::connect error -:" << ec;
             serverError = true;
             break;
           }
         }
 
         boost::asio::streambuf request;
         std::ostream request_stream(&request);
         std::string path = "/popfile?runnumber=" + run_nstring_ + "&pid=" + pid_;
         if (maxLS >= 0) {
           std::stringstream spath;
           spath << path << "&stopls=" << maxLS;
           path = spath.str();
           edm::LogWarning("EvFDaqDirector") << "Stop LS requested " << maxLS;
         }
         request_stream << "GET " << path << " HTTP/1.1\r\n";
         request_stream << "Host: " << fileBrokerHost_ << "\r\n";
         request_stream << "Accept: */*\r\n";
         request_stream << "Connection: keep-alive\r\n\r\n";
 
         boost::asio::write(*socket_, request, ec);
         if (ec) {
           if (fileBrokerKeepAlive_ && ec == boost::asio::error::connection_reset) {
             edm::LogInfo("EvFDaqDirector") << "reconnecting socket on received connection_reset";
             //we got disconnected, try to reconnect to the server before writing the request
             boost::asio::connect(*socket_, *endpoint_iterator_, ec);
             if (ec) {
               edm::LogWarning("EvFDaqDirector") << "boost::asio::connect error -:" << ec;
               serverError = true;
               break;
             }
             continue;
           }
           edm::LogWarning("EvFDaqDirector") << "boost::asio::write error -:" << ec;
           serverError = true;
           break;
         }
 
         boost::asio::streambuf response;
         boost::asio::read_until(*socket_, response, "\r\n", ec);
         if (ec) {
           edm::LogWarning("EvFDaqDirector") << "boost::asio::read_until error -:" << ec;
           serverError = true;
           break;
         }
 
         std::istream response_stream(&response);
 
         std::string http_version;
         response_stream >> http_version;
 
         response_stream >> serverHttpStatus;
 
         std::string status_message;
         std::getline(response_stream, status_message);
         if (!response_stream || http_version.substr(0, 5) != "HTTP/") {
           edm::LogWarning("EvFDaqDirector") << "Invalid server response";
           serverError = true;
           break;
         }
         if (serverHttpStatus != 200) {
           edm::LogWarning("EvFDaqDirector") << "Response returned with status code " << serverHttpStatus;
           serverError = true;
           break;
         }
 
         // Process the response headers.
         std::string header;
         while (std::getline(response_stream, header) && header != "\r") {
         }
 
         std::string fileInfo;
         std::map<std::string, std::string> serverMap;
         while (std::getline(response_stream, fileInfo) && fileInfo != "\r") {
           auto pos = fileInfo.find('=');
           if (pos == std::string::npos)
             continue;
           auto stitle = fileInfo.substr(0, pos);
           auto svalue = fileInfo.substr(pos + 1);
           serverMap[stitle] = svalue;
         }
 
         //check that response run number if correct
         auto server_version = serverMap.find("version");
         assert(server_version != serverMap.end());
 
         auto server_run = serverMap.find("runnumber");
         assert(server_run != serverMap.end());
         assert(run_nstring_ == server_run->second);
 
         auto server_state = serverMap.find("state");
         assert(server_state != serverMap.end());
 
         auto server_eols = serverMap.find("lasteols");
         assert(server_eols != serverMap.end());
 
         auto server_ls = serverMap.find("lumisection");
 
         int version_maj = 1;
         int version_min = 0;
         int version_rev = 0;
         {
           auto* s_ptr = server_version->second.c_str();
           if (!server_version->second.empty() && server_version->second[0] == '"')
             s_ptr++;
           auto res = sscanf(s_ptr, "%d.%d.%d", &version_maj, &version_min, &version_rev);
           if (res < 3) {
             res = sscanf(s_ptr, "%d.%d", &version_maj, &version_min);
             if (res < 2) {
               res = sscanf(s_ptr, "%d", &version_maj);
               if (res < 1) {
                 //expecting at least 1 number (major version)
                 edm::LogWarning("EvFDaqDirector") << "Can not parse server version " << server_version->second;
               }
             }
           }
         }
 
         closedServerLS = (uint64_t)std::max(0, atoi(server_eols->second.c_str()));
         if (server_ls != serverMap.end())
           serverLS = (uint64_t)std::max(1, atoi(server_ls->second.c_str()));
         else
           serverLS = closedServerLS + 1;
 
         std::string s_state = server_state->second;
         if (s_state == "STARTING")  //initial, always empty starting with LS 1
         {
           auto server_file = serverMap.find("file");
           assert(server_file == serverMap.end());  //no file with starting state
           fileStatus = noFile;
           edm::LogInfo("EvFDaqDirector") << "Got STARTING notification with last EOLS " << closedServerLS;
         } else if (s_state == "READY") {
           auto server_file = serverMap.find("file");
           if (server_file == serverMap.end()) {
             //can be returned by server if files from new LS already appeared but LS is not yet closed
             if (serverLS <= closedServerLS)
               serverLS = closedServerLS + 1;
             fileStatus = noFile;
             edm::LogInfo("EvFDaqDirector")
                 << "Got READY notification with last EOLS " << closedServerLS << " and no new file";
           } else {
             std::string filestem;
             std::string fileprefix;
             auto server_fileprefix = serverMap.find("fileprefix");
 
             if (server_fileprefix != serverMap.end()) {
               auto pssize = server_fileprefix->second.size();
               if (pssize > 1 && server_fileprefix->second[0] == '"' && server_fileprefix->second[pssize - 1] == '"')
                 fileprefix = server_fileprefix->second.substr(1, pssize - 2);
               else
                 fileprefix = server_fileprefix->second;
             }
 
             //remove string literals
             auto ssize = server_file->second.size();
             if (ssize > 1 && server_file->second[0] == '"' && server_file->second[ssize - 1] == '"')
               filestem = server_file->second.substr(1, ssize - 2);
             else
               filestem = server_file->second;
             assert(!filestem.empty());
             if (version_maj > 1) {
               nextFileRaw = bu_run_dir_ + "/" + fileprefix + filestem + ".raw";  //filestem should be raw
               filestem = bu_run_dir_ + "/" + fileprefix + filestem;
               nextFileJson = "";
               rawHeader = true;
             } else {
               nextFileRaw = bu_run_dir_ + "/" + filestem + ".raw";  //raw files are not moved
               filestem = bu_run_dir_ + "/" + fileprefix + filestem;
               nextFileJson = filestem + ".jsn";
               rawHeader = false;
             }
             fileStatus = newFile;
             edm::LogInfo("EvFDaqDirector") << "Got READY notification with last EOLS " << closedServerLS << " new LS "
                                            << serverLS << " file:" << filestem;
           }
         } else if (s_state == "EOLS") {
           serverLS = closedServerLS + 1;
           edm::LogInfo("EvFDaqDirector") << "Got EOLS notification with last EOLS " << closedServerLS;
           fileStatus = noFile;
         } else if (s_state == "EOR") {
           //server_eor = serverMap.find("iseor");
           edm::LogInfo("EvFDaqDirector") << "Got EOR notification with last EOLS " << closedServerLS;
           fileStatus = runEnded;
         } else if (s_state == "NORUN") {
           auto err_msg = serverMap.find("errormessage");
           if (err_msg != serverMap.end())
             edm::LogWarning("EvFDaqDirector") << "Server NORUN -:" << server_state->second << " : " << err_msg->second;
           else
             edm::LogWarning("EvFDaqDirector") << "Server NORUN ";
           edm::LogWarning("EvFDaqDirector") << "executing run end";
           fileStatus = runEnded;
         } else if (s_state == "ERROR") {
           auto err_msg = serverMap.find("errormessage");
           if (err_msg != serverMap.end())
             edm::LogWarning("EvFDaqDirector") << "Server error -:" << server_state->second << " : " << err_msg->second;
           else
             edm::LogWarning("EvFDaqDirector") << "Server error -:" << server_state->second;
           fileStatus = noFile;
           serverError = true;
         } else {
           edm::LogWarning("EvFDaqDirector") << "Unknown Server state -:" << server_state->second;
           fileStatus = noFile;
           serverError = true;
         }
 
         // Read until EOF, writing data to output as we go.
         if (!fileBrokerKeepAlive_) {
           while (boost::asio::read(*socket_, response, boost::asio::transfer_at_least(1), ec)) {
           }
           if (ec != boost::asio::error::eof) {
             edm::LogWarning("EvFDaqDirector") << "boost::asio::read_until error -:" << ec;
             serverError = true;
           }
         }
 
         break;
       }
 
     } catch (std::exception const& e) {
       edm::LogWarning("EvFDaqDirector") << "Exception in socket handling";
       serverError = true;
     }
 
     if (!fileBrokerKeepAlive_ && socket_->is_open()) {
       socket_->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
       if (ec) {
         edm::LogWarning("EvFDaqDirector") << "socket shutdown error -:" << ec;
       }
       socket_->close(ec);
       if (ec) {
         edm::LogWarning("EvFDaqDirector") << "socket close error -:" << ec;
       }
     }
 
     if (serverError) {
       if (socket_->is_open())
         socket_->close(ec);
       if (ec) {
         edm::LogWarning("EvFDaqDirector") << "socket close error -:" << ec;
       }
       fileStatus = noFile;
       sleep(1);  //back-off if error detected
     }
 
     return fileStatus;
   }

◆ createBoLSFile()

void evf::EvFDaqDirector::createBoLSFile	(	const uint32_t	lumiSection,
		bool	checkIfExists
	)		const

Definition at line 933 of file EvFDaqDirector.cc.

References visDQMUpload::buf, getBoLSFilePathOnFU(), edm_modernize_messagelogger::stat, and AlCaHLTBitMon_QueryRunRegistry::string.

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

                                                                                           {
     //used for backpressure mechanisms and monitoring
     const std::string fuBoLS = getBoLSFilePathOnFU(lumiSection);
     struct stat buf;
     if (checkIfExists == false || stat(fuBoLS.c_str(), &buf) != 0) {
       int bol_fd = open(fuBoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
       close(bol_fd);
     }
   }

◆ createLumiSectionFiles()

void evf::EvFDaqDirector::createLumiSectionFiles	(	const uint32_t	lumiSection,
		const uint32_t	currentLumiSection,
		bool	doCreateBoLS,
		bool	doCreateEoLS
	)

Definition at line 943 of file EvFDaqDirector.cc.

References visDQMUpload::buf, createBoLSFile(), newFWLiteAna::found, getEoLSFilePathOnFU(), edm_modernize_messagelogger::stat, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by getNextFromFileBroker().

                                                                  {
     if (currentLumiSection > 0) {
       const std::string fuEoLS = getEoLSFilePathOnFU(currentLumiSection);
       struct stat buf;
       bool found = (stat(fuEoLS.c_str(), &buf) == 0);
       if (!found) {
         if (doCreateEoLS) {
           int eol_fd =
               open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
           close(eol_fd);
         }
         if (doCreateBoLS)
           createBoLSFile(lumiSection, false);
       }
     } else if (doCreateBoLS) {
       createBoLSFile(lumiSection, true);  //needed for initial lumisection
     }
   }

◆ createProcessingNotificationMaybe()

void evf::EvFDaqDirector::createProcessingNotificationMaybe ( ) const

Definition at line 2051 of file EvFDaqDirector.cc.

References run_dir_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::checkNext().

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

◆ createRunOpendirMaybe()

void evf::EvFDaqDirector::createRunOpendirMaybe ( )

Definition at line 1872 of file EvFDaqDirector.cc.

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

Referenced by initRun().

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

◆ eolsFileName()

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

private

◆ eorFileName()

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

private

◆ fillDescriptions()

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

static

Definition at line 327 of file EvFDaqDirector.cc.

References edm::ConfigurationDescriptions::add(), submitPVResolutionJobs::desc, 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>("useFileBroker", false)
         ->setComment("Use BU file service to grab input data instead of NFS file locking");
     desc.addUntracked<bool>("fileBrokerHostFromCfg", true)
         ->setComment("Allow service to discover BU address from hltd configuration");
     desc.addUntracked<std::string>("fileBrokerHost", "InValid")->setComment("BU file service host.");
     desc.addUntracked<std::string>("fileBrokerPort", "8080")->setComment("BU file service port");
     desc.addUntracked<bool>("fileBrokerKeepAlive", true)
         ->setComment("Use keep alive to avoid using large number of sockets");
     desc.addUntracked<bool>("fileBrokerUseLocalLock", true)
         ->setComment("Use local lock file to synchronize appearance of index and EoLS file markers for hltd");
     desc.addUntracked<unsigned int>("fuLockPollInterval", 2000)
         ->setComment("Lock polling interval in microseconds for the input directory file lock");
     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<bool>("directorIsBU", false)->setComment("BU director mode used for testing");
     desc.addUntracked<std::string>("hltSourceDirectory", "")->setComment("BU director mode source directory");
     desc.addUntracked<std::string>("mergingPset", "")
         ->setComment("Name of merging PSet to look for merging type definitions for streams");
     descriptions.add("EvFDaqDirector", desc);
   }

◆ findCurrentRunDir()

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

inline

Definition at line 81 of file EvFDaqDirector.h.

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

81 { return dirManager_.findRunDir(run_); }

evf::EvFDaqDirector::dirManager_

DirManager dirManager_

Definition: EvFDaqDirector.h:246

evf::EvFDaqDirector::run_

unsigned int run_

Definition: EvFDaqDirector.h:208

evf::DirManager::findRunDir

std::string findRunDir(unsigned int)

Definition: DirManager.cc:43

◆ getBoLSFilePathOnFU()

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

Definition at line 484 of file EvFDaqDirector.cc.

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

Referenced by createBoLSFile().

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

◆ getDatFilePath()

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

Definition at line 417 of file EvFDaqDirector.cc.

References eostools::ls(), run_, run_dir_, cms::cuda::stream, and fffnaming::streamerDataFileNameWithPid().

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

◆ getEoLSFilePathOnBU()

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

Definition at line 476 of file EvFDaqDirector.cc.

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

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

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

◆ getEoLSFilePathOnFU()

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

Definition at line 480 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::checkNext(), createLumiSectionFiles(), FedRawDataInputSource::maybeOpenNewLumiSection(), and updateFuLock().

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

◆ getEoRFilePath()

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

Definition at line 488 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

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

fffnaming::eorFileName

std::string eorFileName(const unsigned int run)

Definition: FFFNamingSchema.h:34

evf::EvFDaqDirector::run_

unsigned int run_

Definition: EvFDaqDirector.h:208

evf::EvFDaqDirector::bu_run_dir_

std::string bu_run_dir_

Definition: EvFDaqDirector.h:230

◆ getEoRFilePathOnFU()

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

Definition at line 490 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::checkNext().

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

evf::EvFDaqDirector::run_dir_

std::string run_dir_

Definition: EvFDaqDirector.h:229

fffnaming::eorFileName

std::string eorFileName(const unsigned int run)

Definition: FFFNamingSchema.h:34

evf::EvFDaqDirector::run_

unsigned int run_

Definition: EvFDaqDirector.h:208

◆ getFFFParamsFilePathOnBU()

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

Definition at line 492 of file EvFDaqDirector.cc.

References bu_run_dir_.

492 { return bu_run_dir_ + "/hlt/fffParameters.jsn"; }

evf::EvFDaqDirector::bu_run_dir_

std::string bu_run_dir_

Definition: EvFDaqDirector.h:230

◆ getInitFilePath()

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

Definition at line 445 of file EvFDaqDirector.cc.

References fffnaming::initFileNameWithPid(), run_, run_dir_, and cms::cuda::stream.

Referenced by dqm::DQMFileSaverPB::initRun().

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

◆ getInputJsonFilePath()

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

Definition at line 401 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

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

◆ getLumisectionToStart()

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

Definition at line 1896 of file EvFDaqDirector.cc.

References visDQMUpload::buf, reco_skim_cfg_mod::fullpath, run_dir_, run_string_, contentValuesCheck::ss, edm_modernize_messagelogger::stat, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::readSupervisor().

                                                            {
     std::string fileprefix = run_dir_ + "/" + run_string_ + "_ls";
     std::string fullpath;
     struct stat buf;
     unsigned int lscount = 1;
     do {
       std::stringstream ss;
       ss << fileprefix << std::setfill('0') << std::setw(4) << lscount << "_EoLS.jsn";
       fullpath = ss.str();
       lscount++;
     } while (stat(fullpath.c_str(), &buf) == 0);
     return lscount - 1;
   }

◆ getMergedDatChecksumFilePath()

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

Definition at line 437 of file EvFDaqDirector.cc.

References hostname_, eostools::ls(), run_, run_dir_, cms::cuda::stream, and fffnaming::streamerDataChecksumFileNameWithInstance().

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

◆ getMergedDatFilePath()

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

Definition at line 433 of file EvFDaqDirector.cc.

References hostname_, eostools::ls(), run_, run_dir_, cms::cuda::stream, and fffnaming::streamerDataFileNameWithInstance().

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

◆ getMergedProtocolBufferHistogramFilePath()

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

Definition at line 459 of file EvFDaqDirector.cc.

References hostname_, eostools::ls(), fffnaming::protocolBufferHistogramFileNameWithInstance(), run_, run_dir_, and cms::cuda::stream.

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

◆ getMergedRootHistogramFilePath()

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

Definition at line 472 of file EvFDaqDirector.cc.

References hostname_, eostools::ls(), fffnaming::rootHistogramFileNameWithInstance(), run_, run_dir_, and cms::cuda::stream.

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

◆ getNextFromFileBroker()

EvFDaqDirector::FileStatus evf::EvFDaqDirector::getNextFromFileBroker	(	const unsigned int	currentLumiSection,
		unsigned int &	ls,
		std::string &	nextFile,
		int &	rawFd,
		uint16_t &	rawHeaderSize,
		int32_t &	serverEventsInNewFile_,
		int64_t &	fileSize,
		uint64_t &	thisLockWaitTimeUs
	)

Definition at line 1723 of file EvFDaqDirector.cc.

References cms::cuda::assert(), bu_run_dir_, visDQMUpload::buf, contactFileBroker(), createLumiSectionFiles(), fileBrokerUseLocalLock_, grabNextJsonFile(), grabNextJsonFromRaw(), mps_fire::i, lockFULocal(), lockFULocal2(), eostools::ls(), SiStripPI::max, newFile, noFile, readLastLSEntry(), runEnded, edm_modernize_messagelogger::stat, stop_ls_override_, stopFilePath_, stopFilePathPid_, AlCaHLTBitMon_QueryRunRegistry::string, mitigatedMETSequence_cff::U, unlockFULocal(), and unlockFULocal2().

Referenced by FedRawDataInputSource::readSupervisor().

                                                                                                  {
     EvFDaqDirector::FileStatus fileStatus = noFile;
 
     //int retval = -1;
     //int lock_attempts = 0;
     //long total_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;
 
     /* look for EoLS
     if (stat(getEoLSFilePathOnFU(currentLumiSection).c_str(),&buf)==0) {
       edm::LogWarning("EvFDaqDirector") << "Detected local EoLS for lumisection "<< currentLumiSection ; 
       ls++;
       return noFile;
     }
     */
 
     timeval ts_lockbegin;
     gettimeofday(&ts_lockbegin, nullptr);
 
     std::string nextFileJson;
     uint32_t serverLS, closedServerLS;
     unsigned int serverHttpStatus;
     bool serverError;
 
     //local lock to force index json and EoLS files to appear in order
     if (fileBrokerUseLocalLock_)
       lockFULocal();
 
     int maxLS = stopFileLS < 0 ? -1 : std::max(stopFileLS, (int)currentLumiSection);
     bool rawHeader = false;
     fileStatus = contactFileBroker(
         serverHttpStatus, serverError, serverLS, closedServerLS, nextFileJson, nextFileRaw, rawHeader, maxLS);
 
     if (serverError) {
       //do not update anything
       if (fileBrokerUseLocalLock_)
         unlockFULocal();
       return noFile;
     }
 
     //handle creation of BoLS files if lumisection has changed
     if (currentLumiSection == 0) {
       if (fileStatus == runEnded)
         createLumiSectionFiles(closedServerLS, 0, true, false);
       else
         createLumiSectionFiles(serverLS, 0, true, false);
     } else {
       if (closedServerLS >= currentLumiSection) {
         //only BoLS files
         for (uint32_t i = std::max(currentLumiSection, 1U); i <= closedServerLS; i++)
           createLumiSectionFiles(i + 1, i, true, false);
       }
     }
 
     bool fileFound = true;
 
     if (fileStatus == newFile) {
       if (rawHeader > 0)
         serverEventsInNewFile =
             grabNextJsonFromRaw(nextFileRaw, rawFd, rawHeaderSize, fileSizeFromMetadata, fileFound, serverLS, false);
       else
         serverEventsInNewFile = grabNextJsonFile(nextFileJson, nextFileRaw, fileSizeFromMetadata, fileFound);
     }
     //closing file in case of any error
     if (serverEventsInNewFile < 0 && rawFd != -1) {
       close(rawFd);
       rawFd = -1;
     }
 
     //can unlock because all files have been created locally
     if (fileBrokerUseLocalLock_)
       unlockFULocal();
 
     if (!fileFound) {
       //catch condition where directory got deleted
       fileStatus = noFile;
       struct stat buf;
       if (stat(bu_run_dir_.c_str(), &buf) != 0) {
         edm::LogWarning("EvFDaqDirector") << "BU run directory not found:" << bu_run_dir_;
         fileStatus = runEnded;
       }
     }
 
     //handle creation of EoLS files if lumisection has changed, this needs to be locked exclusively
     //so that EoLS files can not appear locally before index files
     if (currentLumiSection == 0) {
       lockFULocal2();
       if (fileStatus == runEnded) {
         createLumiSectionFiles(closedServerLS, 0, false, true);
         createLumiSectionFiles(serverLS, closedServerLS, false, true);  // +1
       } else {
         createLumiSectionFiles(serverLS, 0, false, true);
       }
       unlockFULocal2();
     } else {
       if (closedServerLS >= currentLumiSection) {
         //lock exclusive to create EoLS files
         lockFULocal2();
         for (uint32_t i = std::max(currentLumiSection, 1U); i <= closedServerLS; i++)
           createLumiSectionFiles(i + 1, i, false, true);
         unlockFULocal2();
       }
     }
 
     if (fileStatus == runEnded)
       ls = std::max(currentLumiSection, serverLS);
     else if (fileStatus == newFile) {
       assert(serverLS >= ls);
       ls = serverLS;
     } else if (fileStatus == noFile) {
       if (serverLS >= ls)
         ls = serverLS;
       else {
         edm::LogWarning("EvFDaqDirector") << "Server reported LS " << serverLS
                                           << " which is smaller than currently open LS " << ls << ". Ignoring response";
         sleep(1);
       }
     }
 
     return fileStatus;
   }

◆ getNFilesFromEoLS()

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

private

Definition at line 718 of file EvFDaqDirector.cc.

References bu_run_dir_, visDQMUpload::buf, data, spu::def(), Calorimetry_cff::dp, eolsNFilesIndex_, reco_skim_cfg_mod::fullpath, jsoncollector::DataPointDefinition::getNames(), mps_fire::i, readEolsDefinition_, DQM::reader, edm_modernize_messagelogger::stat, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by bumpFile().

                                                             {
     std::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.empty())
         readEolsDefinition_ = false;
       while (!def.empty()) {
         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().empty()) {
             //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 std::stoi(data);
   }

◆ getOpenDatFilePath()

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

Definition at line 421 of file EvFDaqDirector.cc.

References eostools::ls(), run_, run_dir_, cms::cuda::stream, and fffnaming::streamerDataFileNameWithPid().

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

◆ getOpenInitFilePath()

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

Definition at line 441 of file EvFDaqDirector.cc.

References fffnaming::initFileNameWithPid(), run_, run_dir_, and cms::cuda::stream.

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

◆ getOpenInputJsonFilePath()

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

Definition at line 413 of file EvFDaqDirector.cc.

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

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

◆ getOpenOutputJsonFilePath()

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

Definition at line 425 of file EvFDaqDirector.cc.

References eostools::ls(), run_, run_dir_, cms::cuda::stream, and fffnaming::streamerJsonFileNameWithPid().

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

◆ getOpenProtocolBufferHistogramFilePath()

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

Definition at line 449 of file EvFDaqDirector.cc.

References eostools::ls(), fffnaming::protocolBufferHistogramFileNameWithPid(), run_, run_dir_, and cms::cuda::stream.

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

◆ getOpenRawFilePath()

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

Definition at line 409 of file EvFDaqDirector.cc.

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

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

◆ getOpenRootHistogramFilePath()

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

Definition at line 464 of file EvFDaqDirector.cc.

References eostools::ls(), fffnaming::rootHistogramFileNameWithPid(), run_, run_dir_, and cms::cuda::stream.

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

◆ getOutputJsonFilePath()

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

Definition at line 429 of file EvFDaqDirector.cc.

References eostools::ls(), run_, run_dir_, cms::cuda::stream, and fffnaming::streamerJsonFileNameWithPid().

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

◆ getProtocolBufferHistogramFilePath()

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

Definition at line 454 of file EvFDaqDirector.cc.

References eostools::ls(), fffnaming::protocolBufferHistogramFileNameWithPid(), run_, run_dir_, and cms::cuda::stream.

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

◆ getRawFilePath()

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

Definition at line 405 of file EvFDaqDirector.cc.

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

Referenced by bumpFile(), and removeFile().

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

◆ getRootHistogramFilePath()

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

Definition at line 468 of file EvFDaqDirector.cc.

References eostools::ls(), fffnaming::rootHistogramFileNameWithPid(), run_, run_dir_, and cms::cuda::stream.

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

◆ getRunNumber()

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

inline

Definition at line 118 of file EvFDaqDirector.h.

References run_.

118 { return run_; }

evf::EvFDaqDirector::run_

unsigned int run_

Definition: EvFDaqDirector.h:208

◆ getRunOpenDirPath()

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

inline

Definition at line 106 of file EvFDaqDirector.h.

References run_dir_.

Referenced by createRunOpendirMaybe(), and initRun().

106 { return run_dir_ + "/open"; }

evf::EvFDaqDirector::run_dir_

std::string run_dir_

Definition: EvFDaqDirector.h:229

◆ getStartLumisectionFromEnv()

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

inline

Definition at line 176 of file EvFDaqDirector.h.

References startFromLS_.

Referenced by FedRawDataInputSource::readSupervisor().

176 { return startFromLS_; }

evf::EvFDaqDirector::startFromLS_

unsigned int startFromLS_

Definition: EvFDaqDirector.h:223

◆ getStreamDestinations()

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

Definition at line 1973 of file EvFDaqDirector.cc.

References Json::Value::begin(), Json::Value::end(), Exception, mps_check::msg, requireTSPSet_, runTheMatrix::ret, selectedTransferMode_, cms::cuda::stream, 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_.empty() || 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_.empty() || 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.empty())
         ret += ",";
       ret += (*it).asString();
     }
     return ret;
   }

◆ getStreamMergeType()

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

Definition at line 2037 of file EvFDaqDirector.cc.

References mergeTypeMap_, MergeTypeNames_, cms::cuda::stream, and AlCaHLTBitMon_QueryRunRegistry::string.

                                                                                              {
     tbb::concurrent_hash_map<std::string, std::string>::const_accessor search_ac;
     if (mergeTypeMap_.find(search_ac, stream))
       return search_ac->second;
 
     edm::LogInfo("EvFDaqDirector") << " No merging type specified for stream " << stream << ". Using default value";
     std::string defaultName = MergeTypeNames_[defaultType];
     tbb::concurrent_hash_map<std::string, std::string>::accessor ac;
     mergeTypeMap_.insert(ac, stream);
     ac->second = defaultName;
     ac.release();
     return defaultName;
   }

◆ grabNextJsonFile()

int evf::EvFDaqDirector::grabNextJsonFile	(	std::string const &	jsonSourcePath,
		std::string const &	rawSourcePath,
		int64_t &	fileSizeFromJson,
		bool &	fileFound
	)

Definition at line 1182 of file EvFDaqDirector.cc.

References cms::cuda::assert(), baseRunDir(), visDQMUpload::buf, data, Calorimetry_cff::dp, dpd_, MillePedeFileConverter_cfg::e, cppFunctionSkipper::exception, jsoncollector::DataPointDefinition::getNames(), mps_fire::i, timingPdfMaker::infile, LogDebug, timingPdfMaker::outfile, castor_dqm_sourceclient_file_cfg::path, pid_, fileinputsource_cfi::read, DQM::reader, mps_fire::result, contentValuesCheck::ss, edm_modernize_messagelogger::stat, AlCaHLTBitMon_QueryRunRegistry::string, summarizeEdmComparisonLogfiles::success, and writeEcalDQMStatus::write.

Referenced by getNextFromFileBroker().

                                                         {
     fileFound = true;
 
     //should be ported to use fffnaming
     std::ostringstream fileNameWithPID;
     fileNameWithPID << std::filesystem::path(rawSourcePath).stem().string() << "_pid" << std::setfill('0')
                     << std::setw(5) << pid_ << ".jsn";
 
     // assemble json destination path
     std::string jsonDestPath = baseRunDir() + "/" + fileNameWithPID.str();
 
     LogDebug("EvFDaqDirector") << "JSON rename -: " << jsonSourcePath << " to " << jsonDestPath;
 
     int infile = -1, outfile = -1;
 
     if ((infile = ::open(jsonSourcePath.c_str(), O_RDONLY)) < 0) {
       edm::LogWarning("EvFDaqDirector") << "grabNextJsonFile - failed to open input file -: " << jsonSourcePath << " : "
                                         << strerror(errno);
       if ((infile = ::open(jsonSourcePath.c_str(), O_RDONLY)) < 0) {
         edm::LogError("EvFDaqDirector") << "grabNextJsonFile - failed to open input file (on retry) -: "
                                         << jsonSourcePath << " : " << strerror(errno);
         if (errno == ENOENT)
           fileFound = false;
         return -1;
       }
     }
 
     int oflag = O_CREAT | O_WRONLY | O_TRUNC | O_EXCL;  //file should not exist
     int omode = S_IWUSR | S_IRUSR | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH;
     if ((outfile = ::open(jsonDestPath.c_str(), oflag, omode)) < 0) {
       if (errno == EEXIST) {
         edm::LogError("EvFDaqDirector") << "grabNextJsonFile - destination file already exists -: " << jsonDestPath
                                         << " : ";
         ::close(infile);
         return -1;
       }
       edm::LogError("EvFDaqDirector") << "grabNextJsonFile - failed to open output file -: " << jsonDestPath << " : "
                                       << strerror(errno);
       struct stat out_stat;
       if (stat(jsonDestPath.c_str(), &out_stat) == 0) {
         edm::LogWarning("EvFDaqDirector")
             << "grabNextJsonFile - output file possibly got created with error, deleting and retry -: " << jsonDestPath;
         if (unlink(jsonDestPath.c_str()) == -1) {
           edm::LogWarning("EvFDaqDirector")
               << "grabNextJsonFile - failed to remove -: " << jsonDestPath << " : " << strerror(errno);
         }
       }
       if ((outfile = ::open(jsonDestPath.c_str(), oflag, omode)) < 0) {
         edm::LogError("EvFDaqDirector") << "grabNextJsonFile - failed to open output file (on retry) -: "
                                         << jsonDestPath << " : " << strerror(errno);
         ::close(infile);
         return -1;
       }
     }
     //copy contents
     const std::size_t buf_sz = 512;
     std::size_t tot_written = 0;
     std::unique_ptr<char> buf(new char[buf_sz]);
 
     ssize_t sz, sz_read = 1, sz_write;
     while (sz_read > 0 && (sz_read = ::read(infile, buf.get(), buf_sz)) > 0) {
       sz_write = 0;
       do {
         assert(sz_read - sz_write > 0);
         if ((sz = ::write(outfile, buf.get() + sz_write, sz_read - sz_write)) < 0) {
           sz_read = sz;  // cause read loop termination
           break;
         }
         assert(sz > 0);
         sz_write += sz;
         tot_written += sz;
       } while (sz_write < sz_read);
     }
     close(infile);
     close(outfile);
 
     if (tot_written > 0) {
       //leave file if it was empty for diagnosis
       if (unlink(jsonSourcePath.c_str()) == -1) {
         edm::LogError("EvFDaqDirector") << "grabNextJsonFile - failed to remove -: " << jsonSourcePath << " : "
                                         << strerror(errno);
         return -1;
       }
     } else {
       edm::LogError("EvFDaqDirector") << "grabNextJsonFile - failed to copy json file or file was empty -: "
                                       << jsonSourcePath;
       return -1;
     }
 
     Json::Value deserializeRoot;
     Json::Reader reader;
 
     std::string data;
     std::stringstream ss;
     bool result;
     try {
       if (tot_written <= buf_sz) {
         result = reader.parse(buf.get(), deserializeRoot);
       } else {
         //json will normally not be bigger than buf_sz bytes
         try {
           std::ifstream ij(jsonDestPath);
           ss << ij.rdbuf();
         } catch (std::filesystem::filesystem_error const& ex) {
           edm::LogError("EvFDaqDirector") << "grabNextJsonFile - FILESYSTEM ERROR CAUGHT -: " << ex.what();
           return -1;
         }
         result = reader.parse(ss.str(), deserializeRoot);
       }
       if (!result) {
         if (tot_written <= buf_sz)
           ss << buf.get();
         edm::LogError("EvFDaqDirector") << "Failed to deserialize JSON file -: " << jsonDestPath << "\nERROR:\n"
                                         << reader.getFormatedErrorMessages() << "CONTENT:\n"
                                         << ss.str() << ".";
         return -1;
       }
 
       //read BU JSON
       DataPoint dp;
       dp.deserialize(deserializeRoot);
       bool success = false;
       for (unsigned int i = 0; i < dpd_->getNames().size(); i++) {
         if (dpd_->getNames().at(i) == "NEvents")
           if (i < dp.getData().size()) {
             data = dp.getData()[i];
             success = true;
             break;
           }
       }
       if (!success) {
         if (!dp.getData().empty())
           data = dp.getData()[0];
         else {
           edm::LogError("EvFDaqDirector::grabNextJsonFile")
               << "grabNextJsonFile - "
               << " error reading number of events from BU JSON; No input value. data -: " << data;
           return -1;
         }
       }
 
       //try to read raw file size
       fileSizeFromJson = -1;
       for (unsigned int i = 0; i < dpd_->getNames().size(); i++) {
         if (dpd_->getNames().at(i) == "NBytes") {
           if (i < dp.getData().size()) {
             std::string dataSize = dp.getData()[i];
             try {
               fileSizeFromJson = std::stol(dataSize);
             } catch (const std::exception&) {
               //non-fatal currently, processing can continue without this value
               edm::LogWarning("EvFDaqDirector") << "grabNextJsonFile - error parsing number of Bytes from BU JSON. "
                                                 << "Input value is -: " << dataSize;
             }
             break;
           }
         }
       }
       return std::stoi(data);
     } catch (const std::out_of_range& e) {
       edm::LogError("EvFDaqDirector") << "grabNextJsonFile - error parsing number of events from BU JSON. "
                                       << "Input value is -: " << data;
     } catch (const std::invalid_argument& e) {
       edm::LogError("EvFDaqDirector") << "grabNextJsonFile - argument error parsing events from BU JSON. "
                                       << "Input value is -: " << data;
     } catch (std::runtime_error const& e) {
       //Can be thrown by Json parser
       edm::LogError("EvFDaqDirector") << "grabNextJsonFile - std::runtime_error exception -: " << e.what();
     }
 
     catch (std::exception const& e) {
       edm::LogError("EvFDaqDirector") << "grabNextJsonFile - SOME OTHER EXCEPTION OCCURED! -: " << e.what();
     } catch (...) {
       //unknown exception
       edm::LogError("EvFDaqDirector") << "grabNextJsonFile - SOME OTHER EXCEPTION OCCURED!";
     }
 
     return -1;
   }

◆ grabNextJsonFileAndUnlock()

int evf::EvFDaqDirector::grabNextJsonFileAndUnlock ( std::filesystem::path const & jsonSourcePath )

Definition at line 1365 of file EvFDaqDirector.cc.

References baseRunDir(), filterCSVwithJSON::copy, data, Calorimetry_cff::dp, dpd_, MillePedeFileConverter_cfg::e, cppFunctionSkipper::exception, Exception, jsoncollector::DataPointDefinition::getNames(), mps_fire::i, LogDebug, castor_dqm_sourceclient_file_cfg::path, DQM::reader, MatrixUtil::remove(), contentValuesCheck::ss, AlCaHLTBitMon_QueryRunRegistry::string, summarizeEdmComparisonLogfiles::success, and unlockFULocal().

Referenced by FedRawDataInputSource::readSupervisor().

                                                                                        {
     std::string data;
     try {
       // assemble json destination path
       std::filesystem::path jsonDestPath(baseRunDir());
 
       //should be ported to use fffnaming
       std::ostringstream fileNameWithPID;
       fileNameWithPID << jsonSourcePath.stem().string() << "_pid" << std::setfill('0') << std::setw(5) << getpid()
                       << ".jsn";
       jsonDestPath /= fileNameWithPID.str();
 
       LogDebug("EvFDaqDirector") << "JSON rename -: " << jsonSourcePath << " to " << jsonDestPath;
       try {
         std::filesystem::copy(jsonSourcePath, jsonDestPath);
       } catch (std::filesystem::filesystem_error const& ex) {
         // Input dir gone?
         edm::LogError("EvFDaqDirector") << "grabNextFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what();
         //                                     << " Maybe the file is not yet visible by FU. Trying again in one second";
         sleep(1);
         std::filesystem::copy(jsonSourcePath, jsonDestPath);
       }
       unlockFULocal();
 
       try {
         //sometimes this fails but file gets deleted
         std::filesystem::remove(jsonSourcePath);
       } catch (std::filesystem::filesystem_error const& ex) {
         // Input dir gone?
         edm::LogError("EvFDaqDirector") << "grabNextFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what();
       } catch (std::exception const& ex) {
         // Input dir gone?
         edm::LogError("EvFDaqDirector") << "grabNextFile std::exception CAUGHT -: " << ex.what();
       }
 
       std::ifstream ij(jsonDestPath);
       Json::Value deserializeRoot;
       Json::Reader reader;
 
       std::stringstream ss;
       ss << ij.rdbuf();
       if (!reader.parse(ss.str(), deserializeRoot)) {
         edm::LogError("EvFDaqDirector") << "grabNextFile Failed to deserialize JSON file -: " << jsonDestPath
                                         << "\nERROR:\n"
                                         << reader.getFormatedErrorMessages() << "CONTENT:\n"
                                         << ss.str() << ".";
         throw std::runtime_error("Cannot deserialize input JSON file");
       }
 
       //read BU JSON
       std::string data;
       DataPoint dp;
       dp.deserialize(deserializeRoot);
       bool success = false;
       for (unsigned int i = 0; i < dpd_->getNames().size(); i++) {
         if (dpd_->getNames().at(i) == "NEvents")
           if (i < dp.getData().size()) {
             data = dp.getData()[i];
             success = true;
           }
       }
       if (!success) {
         if (!dp.getData().empty())
           data = dp.getData()[0];
         else
           throw cms::Exception("EvFDaqDirector::grabNextJsonFileUnlock")
               << " error reading number of events from BU JSON -: No input value " << data;
       }
       return std::stoi(data);
     } catch (std::filesystem::filesystem_error const& ex) {
       // Input dir gone?
       unlockFULocal();
       edm::LogError("EvFDaqDirector") << "grabNextFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what();
     } catch (std::runtime_error const& e) {
       // Another process grabbed the file and NFS did not register this
       unlockFULocal();
       edm::LogError("EvFDaqDirector") << "grabNextFile runtime Exception -: " << e.what();
     } catch (const std::out_of_range&) {
       edm::LogError("EvFDaqDirector") << "grabNextFile error parsing number of events from BU JSON. "
                                       << "Input value is -: " << data;
     } catch (const std::invalid_argument&) {
       edm::LogError("EvFDaqDirector") << "grabNextFile argument error parsing events from BU JSON. "
                                       << "Input value is -: " << data;
     } catch (std::exception const& e) {
       // BU run directory disappeared?
       unlockFULocal();
       edm::LogError("EvFDaqDirector") << "grabNextFile SOME OTHER EXCEPTION OCCURED!!!! -: " << e.what();
     }
 
     return -1;
   }

◆ grabNextJsonFromRaw()

int evf::EvFDaqDirector::grabNextJsonFromRaw	(	std::string const &	rawSourcePath,
		int &	rawFd,
		uint16_t &	rawHeaderSize,
		int64_t &	fileSizeFromHeader,
		bool &	fileFound,
		uint32_t	serverLS,
		bool	closeFile
	)

Definition at line 1100 of file EvFDaqDirector.cc.

References baseRunDir(), LogDebug, timingPdfMaker::outfile, parseFRDFileHeader(), castor_dqm_sourceclient_file_cfg::path, pid_, runTheMatrix::ret, contentValuesCheck::ss, edm_modernize_messagelogger::stat, AlCaHLTBitMon_QueryRunRegistry::string, and writeEcalDQMStatus::write.

Referenced by getNextFromFileBroker(), and FedRawDataInputSource::readSupervisor().

                                                           {
     fileFound = true;
 
     //take only first three tokens delimited by "_" in the renamed raw file name
     std::string jsonStem = std::filesystem::path(rawSourcePath).stem().string();
     size_t pos = 0, n_tokens = 0;
     while (n_tokens++ < 3 && (pos = jsonStem.find('_', pos + 1)) != std::string::npos) {
     }
     std::string reducedJsonStem = jsonStem.substr(0, pos);
 
     std::ostringstream fileNameWithPID;
     //should be ported to use fffnaming
     fileNameWithPID << reducedJsonStem << "_pid" << std::setfill('0') << std::setw(5) << pid_ << ".jsn";
 
     std::string jsonDestPath = baseRunDir() + "/" + fileNameWithPID.str();
 
     LogDebug("EvFDaqDirector") << "RAW parse -: " << rawSourcePath << " and JSON create " << jsonDestPath;
 
     //parse RAW file header if it exists
     uint32_t lsFromRaw;
     int32_t nbEventsWrittenRaw;
     int64_t fileSizeFromRaw;
     auto ret = parseFRDFileHeader(
         rawSourcePath, rawFd, rawHeaderSize, lsFromRaw, nbEventsWrittenRaw, fileSizeFromRaw, true, true, closeFile);
     if (ret != 0) {
       if (ret == 1)
         fileFound = false;
       return -1;
     }
 
     int outfile;
     int oflag = O_CREAT | O_WRONLY | O_TRUNC | O_EXCL;  //file should not exist
     int omode = S_IWUSR | S_IRUSR | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH;
     if ((outfile = ::open(jsonDestPath.c_str(), oflag, omode)) < 0) {
       if (errno == EEXIST) {
         edm::LogError("EvFDaqDirector") << "grabNextJsonFromRaw - destination file already exists -: " << jsonDestPath
                                         << " : ";
         return -1;
       }
       edm::LogError("EvFDaqDirector") << "grabNextJsonFromRaw - failed to open output file -: " << jsonDestPath << " : "
                                       << strerror(errno);
       struct stat out_stat;
       if (stat(jsonDestPath.c_str(), &out_stat) == 0) {
         edm::LogWarning("EvFDaqDirector")
             << "grabNextJsonFromRaw - output file possibly got created with error, deleting and retry -: "
             << jsonDestPath;
         if (unlink(jsonDestPath.c_str()) == -1) {
           edm::LogWarning("EvFDaqDirector")
               << "grabNextJsonFromRaw - failed to remove -: " << jsonDestPath << " : " << strerror(errno);
         }
       }
       if ((outfile = ::open(jsonDestPath.c_str(), oflag, omode)) < 0) {
         edm::LogError("EvFDaqDirector") << "grabNextJsonFromRaw - failed to open output file (on retry) -: "
                                         << jsonDestPath << " : " << strerror(errno);
         return -1;
       }
     }
     //write JSON file (TODO: use jsoncpp)
     std::stringstream ss;
     ss << "{\"data\":[" << nbEventsWrittenRaw << "," << fileSizeFromRaw << ",\"" << rawSourcePath << "\"]}";
     std::string sstr = ss.str();
 
     if (::write(outfile, sstr.c_str(), sstr.size()) < 0) {
       edm::LogError("EvFDaqDirector") << "grabNextJsonFromRaw - failed to write to output file file -: " << jsonDestPath
                                       << " : " << strerror(errno);
       return -1;
     }
     close(outfile);
     if (serverLS && serverLS != lsFromRaw)
       edm::LogWarning("EvFDaqDirector") << "grabNextJsonFromRaw - mismatch in expected (server) LS " << serverLS
                                         << " and raw file header LS " << lsFromRaw;
 
     fileSizeFromHeader = fileSizeFromRaw;
     return nbEventsWrittenRaw;
   }

◆ initFileName()

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

private

◆ initRun()

void evf::EvFDaqDirector::initRun ( )

Definition at line 148 of file EvFDaqDirector.cc.

References base_dir_, bu_base_dir_, bu_run_dir_, bu_run_open_dir_, bu_w_lock_stream, bu_writelock_fd_, visDQMUpload::buf, eostools::chmod(), createRunOpendirMaybe(), directorBU_, dpd_, Exception, fu_readwritelock_fd_, fu_rw_lock_stream, fulocal_rwlock_fd2_, fulocal_rwlock_fd_, fulockfile_, getRunOpenDirPath(), hltSourceDirectory_, init_lock_, input_throttled_file_, eostools::mkdir(), openFULockfileStream(), pid_, run_, run_dir_, run_nstring_, run_string_, contentValuesCheck::ss, edm_modernize_messagelogger::stat, 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();
     ss = std::stringstream();
     ss << run_;
     run_nstring_ = ss.str();
     run_dir_ = base_dir_ + "/" + run_string_;
     input_throttled_file_ = run_dir_ + "/input_throttle";
     ss = std::stringstream();
     ss << getpid();
     pid_ = ss.str();
 
     // 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";
       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 == nullptr)
         edm::LogWarning("EvFDaqDirector") << "Error creating write lock stream -: " << strerror(errno);
 
       // BU INITIALIZES LOCK FILE
       // FU LOCK FILE OPEN
       openFULockfileStream(true);
       tryInitializeFuLockFile();
       fflush(fu_rw_lock_stream);
       close(fu_readwritelock_fd_);
 
       if (!hltSourceDirectory_.empty()) {
         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";
 
           std::filesystem::copy_file(hltSourceDirectory_ + "/HltConfig.py", tmphltdir + "/HltConfig.py");
           std::filesystem::copy_file(hltSourceDirectory_ + "/fffParameters.jsn", tmphltdir + "/fffParameters.jsn");
 
           std::string optfiles[3] = {"hltinfo", "blacklist", "whitelist"};
           for (auto& optfile : optfiles) {
             try {
               std::filesystem::copy_file(hltSourceDirectory_ + "/" + optfile, tmphltdir + "/" + optfile);
             } catch (...) {
             }
           }
 
           std::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_;
       fulockfile_ = bu_run_dir_ + "/fu.lock";
       openFULockfileStream(false);
     }
 
     pthread_mutex_init(&init_lock_, nullptr);
 
     stopFilePath_ = run_dir_ + "/CMSSW_STOP";
     std::stringstream sstp;
     sstp << stopFilePath_ << "_pid" << pid_;
     stopFilePathPid_ = sstp.str();
 
     if (!directorBU_) {
       std::string defPath = bu_run_dir_ + "/jsd/rawData.jsd";
       struct stat statbuf;
       if (!stat(defPath.c_str(), &statbuf))
         edm::LogInfo("EvFDaqDirector") << "found JSD file in ramdisk -: " << defPath;
       else {
         //look in source directory if not present in ramdisk
         std::string defPathSuffix = "src/EventFilter/Utilities/plugins/budef.jsd";
         defPath = std::string(std::getenv("CMSSW_BASE")) + "/" + defPathSuffix;
         if (stat(defPath.c_str(), &statbuf)) {
           defPath = std::string(std::getenv("CMSSW_RELEASE_BASE")) + "/" + defPathSuffix;
           if (stat(defPath.c_str(), &statbuf)) {
             defPath = defPathSuffix;
           }
         }
       }
       dpd_ = new DataPointDefinition();
       std::string defLabel = "data";
       DataPointDefinition::getDataPointDefinitionFor(defPath, dpd_, &defLabel);
     }
   }

◆ inputFileNameStem()

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

private

◆ inputThrottled()

bool evf::EvFDaqDirector::inputThrottled ( )

Definition at line 2065 of file EvFDaqDirector.cc.

References visDQMUpload::buf, input_throttled_file_, and edm_modernize_messagelogger::stat.

Referenced by FedRawDataInputSource::readSupervisor().

                                       {
     struct stat buf;
     return (stat(input_throttled_file_.c_str(), &buf) == 0);
   }

◆ isSingleStreamThread()

bool evf::EvFDaqDirector::isSingleStreamThread ( )

inline

Definition at line 122 of file EvFDaqDirector.h.

References nStreams_, and nThreads_.

Referenced by FedRawDataInputSource::getNextEvent().

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

evf::EvFDaqDirector::nThreads_

unsigned int nThreads_

Definition: EvFDaqDirector.h:265

evf::EvFDaqDirector::nStreams_

unsigned int nStreams_

Definition: EvFDaqDirector.h:264

◆ lockFULocal()

void evf::EvFDaqDirector::lockFULocal ( )

Definition at line 913 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by getNextFromFileBroker(), and updateFuLock().

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

◆ lockFULocal2()

void evf::EvFDaqDirector::lockFULocal2 ( )

Definition at line 923 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

Referenced by FedRawDataInputSource::checkNext(), getNextFromFileBroker(), and FedRawDataInputSource::maybeOpenNewLumiSection().

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

◆ lockInitLock()

void evf::EvFDaqDirector::lockInitLock ( )

Definition at line 909 of file EvFDaqDirector.cc.

References init_lock_.

909 { pthread_mutex_lock(&init_lock_); }

evf::EvFDaqDirector::init_lock_

pthread_mutex_t init_lock_

Definition: EvFDaqDirector.h:262

◆ make_flock()

struct flock evf::EvFDaqDirector::make_flock	(	short	type,
		short	whence,
		off_t	start,
		off_t	len,
		pid_t	pid
	)

static

Definition at line 2057 of file EvFDaqDirector.cc.

References command_line::start.

                                                                                                      {
 #ifdef __APPLE__
     return {start, len, pid, type, whence};
 #else
     return {type, whence, start, len, pid};
 #endif
   }

◆ mergedFileNameStem()

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

private

◆ openFULockfileStream()

void evf::EvFDaqDirector::openFULockfileStream ( bool create )

private

Definition at line 890 of file EvFDaqDirector.cc.

References eostools::chmod(), beamerCreator::create(), fu_readwritelock_fd_, fu_rw_lock_stream, fulockfile_, 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_
                                       << " create:" << create << " error:" << strerror(errno);
     else
       LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: " << fu_readwritelock_fd_;
 
     fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
     if (fu_rw_lock_stream == nullptr)
       edm::LogError("EvFDaqDirector") << "problem with opening fuwritelock file stream -: " << strerror(errno);
   }

◆ outputAdler32Recheck()

bool evf::EvFDaqDirector::outputAdler32Recheck ( ) const

inline

Definition at line 107 of file EvFDaqDirector.h.

References outputAdler32Recheck_.

107 { return outputAdler32Recheck_; }

evf::EvFDaqDirector::outputAdler32Recheck_

bool outputAdler32Recheck_

Definition: EvFDaqDirector.h:216

◆ outputFileNameStem()

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

private

◆ overrideRunNumber()

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

inline

Definition at line 75 of file EvFDaqDirector.h.

References writedatasetfile::run, and run_.

75 { run_ = run; }

writedatasetfile.run

run

Definition: writedatasetfile.py:27

evf::EvFDaqDirector::run_

unsigned int run_

Definition: EvFDaqDirector.h:208

◆ parseFRDFileHeader()

int evf::EvFDaqDirector::parseFRDFileHeader	(	std::string const &	rawSourcePath,
		int &	rawFd,
		uint16_t &	rawHeaderSize,
		uint32_t &	lsFromHeader,
		int32_t &	eventsFromHeader,
		int64_t &	fileSizeFromHeader,
		bool	requireHeader,
		bool	retry,
		bool	closeFile
	)

static

Definition at line 965 of file EvFDaqDirector.cc.

References getFRDFileHeaderVersion(), timingPdfMaker::infile, and fileinputsource_cfi::read.

Referenced by grabNextJsonFromRaw(), and FedRawDataInputSource::readSupervisor().

                                                          {
     int infile;
 
     if ((infile = ::open(rawSourcePath.c_str(), O_RDONLY)) < 0) {
       if (retry) {
         edm::LogWarning("EvFDaqDirector")
             << "parseFRDFileHeader - failed to open input file -: " << rawSourcePath << " : " << strerror(errno);
         return parseFRDFileHeader(rawSourcePath,
                                   rawFd,
                                   rawHeaderSize,
                                   lsFromHeader,
                                   eventsFromHeader,
                                   fileSizeFromHeader,
                                   requireHeader,
                                   false,
                                   closeFile);
       } else {
         if ((infile = ::open(rawSourcePath.c_str(), O_RDONLY)) < 0) {
           edm::LogError("EvFDaqDirector")
               << "parseFRDFileHeader - failed to open input file -: " << rawSourcePath << " : " << strerror(errno);
           if (errno == ENOENT)
             return 1;  // error && file not found
           else
             return -1;
         }
       }
     }
 
     constexpr std::size_t buf_sz = sizeof(FRDFileHeader_v1);  //try to read v1 FRD header size
     FRDFileHeader_v1 fileHead;
 
     ssize_t sz_read = ::read(infile, (char*)&fileHead, buf_sz);
     if (closeFile) {
       close(infile);
       infile = -1;
     }
 
     if (sz_read < 0) {
       edm::LogError("EvFDaqDirector") << "parseFRDFileHeader - unable to read " << rawSourcePath << " : "
                                       << strerror(errno);
       if (infile != -1)
         close(infile);
       return -1;
     }
     if ((size_t)sz_read < buf_sz) {
       edm::LogError("EvFDaqDirector") << "parseFRDFileHeader - file smaller than header: " << rawSourcePath;
       if (infile != -1)
         close(infile);
       return -1;
     }
 
     uint16_t frd_version = getFRDFileHeaderVersion(fileHead.id_, fileHead.version_);
 
     if (frd_version == 0) {
       //no header (specific sequence not detected)
       if (requireHeader) {
         edm::LogError("EvFDaqDirector") << "no header or invalid version string found in:" << rawSourcePath;
         if (infile != -1)
           close(infile);
         return -1;
       } else {
         //no header, but valid file
         lseek(infile, 0, SEEK_SET);
         rawHeaderSize = 0;
         lsFromHeader = 0;
         eventsFromHeader = -1;
         fileSizeFromHeader = -1;
       }
     } else {
       //version 1 header
       uint32_t headerSizeRaw = fileHead.headerSize_;
       if (headerSizeRaw < buf_sz) {
         edm::LogError("EvFDaqDirector") << "inconsistent header size: " << rawSourcePath << " size: " << headerSizeRaw
                                         << " v:" << frd_version;
         if (infile != -1)
           close(infile);
         return -1;
       }
       //allow header size to exceed read size. Future header versions will not break this, but the size can change.
       lsFromHeader = fileHead.lumiSection_;
       eventsFromHeader = (int32_t)fileHead.eventCount_;
       fileSizeFromHeader = (int64_t)fileHead.fileSize_;
       rawHeaderSize = fileHead.headerSize_;
     }
     rawFd = infile;
     return 0;  //OK
   }

◆ postEndRun()

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

Definition at line 374 of file EvFDaqDirector.cc.

References bu_readlock_fd_, bu_run_dir_, bu_writelock_fd_, directorBU_, corrVsCorr::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);
     }
   }

◆ preallocate()

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

Definition at line 320 of file EvFDaqDirector.cc.

References initRun(), nStreams_, and nThreads_.

Referenced by EvFDaqDirector().

                                                                        {
     initRun();
 
     nThreads_ = bounds.maxNumberOfStreams();
     nStreams_ = bounds.maxNumberOfThreads();
   }

◆ preBeginJob()

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

Definition at line 359 of file EvFDaqDirector.cc.

References checkMergeTypePSet(), and checkTransferSystemPSet().

Referenced by EvFDaqDirector().

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

◆ preBeginRun()

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

Definition at line 364 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();
     }
   }

◆ preGlobalEndLumi()

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

Definition at line 383 of file EvFDaqDirector.cc.

References fileDeleteLockPtr_, filesToDeletePtr_, fms_, evf::FastMonitoringService::isExceptionOnData(), eostools::ls(), edm::LuminosityBlockID::luminosityBlock(), and edm::GlobalContext::luminosityBlockID().

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 && (!fms_ || !fms_->isExceptionOnData(it->second->lumi_))) {
         it = filesToDeletePtr_->erase(it);
       } else
         it++;
     }
   }

◆ rawFileHasHeader()

bool evf::EvFDaqDirector::rawFileHasHeader	(	std::string const &	rawSourcePath,
		uint16_t &	rawHeaderSize
	)

Definition at line 1061 of file EvFDaqDirector.cc.

References getFRDFileHeaderVersion(), timingPdfMaker::infile, and fileinputsource_cfi::read.

Referenced by bumpFile().

                                                                                                {
     int infile;
     if ((infile = ::open(rawSourcePath.c_str(), O_RDONLY)) < 0) {
       edm::LogWarning("EvFDaqDirector") << "rawFileHasHeader - failed to open input file -: " << rawSourcePath << " : "
                                         << strerror(errno);
       return false;
     }
     constexpr std::size_t buf_sz = sizeof(FRDFileHeader_v1);  //try to read v1 FRD header size
     FRDFileHeader_v1 fileHead;
 
     ssize_t sz_read = ::read(infile, (char*)&fileHead, buf_sz);
 
     if (sz_read < 0) {
       edm::LogError("EvFDaqDirector") << "rawFileHasHeader - unable to read " << rawSourcePath << " : "
                                       << strerror(errno);
       if (infile != -1)
         close(infile);
       return false;
     }
     if ((size_t)sz_read < buf_sz) {
       edm::LogError("EvFDaqDirector") << "rawFileHasHeader - file smaller than header: " << rawSourcePath;
       if (infile != -1)
         close(infile);
       return false;
     }
 
     uint16_t frd_version = getFRDFileHeaderVersion(fileHead.id_, fileHead.version_);
 
     close(infile);
 
     if (frd_version > 0) {
       rawHeaderSize = fileHead.headerSize_;
       return true;
     }
 
     rawHeaderSize = 0;
     return false;
   }

◆ readLastLSEntry()

int evf::EvFDaqDirector::readLastLSEntry ( std::string const & file )

Definition at line 1882 of file EvFDaqDirector.cc.

References Json::Value::asInt(), geometryDiff::file, Json::Value::get(), DQM::reader, and runTheMatrix::ret.

Referenced by getNextFromFileBroker(), and updateFuLock().

                                                            {
     std::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;
   }

◆ removeFile() [1/2]

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

Definition at line 501 of file EvFDaqDirector.cc.

References getRawFilePath(), and eostools::ls().

Referenced by postEndRun().

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

evf::EvFDaqDirector::getRawFilePath

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

Definition: EvFDaqDirector.cc:405

AlignmentPI::index

index

Definition: AlignmentPayloadInspectorHelper.h:46

evf::EvFDaqDirector::removeFile

void removeFile(unsigned int ls, unsigned int index)

Definition: EvFDaqDirector.cc:501

eostools.ls

def ls(path, rec=False)

Definition: eostools.py:349

◆ removeFile() [2/2]

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

Definition at line 494 of file EvFDaqDirector.cc.

References corrVsCorr::filename.

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

◆ setDeleteTracking()

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

inline

Definition at line 177 of file EvFDaqDirector.h.

References fileDeleteLockPtr_, and filesToDeletePtr_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

                                                                                            {
       fileDeleteLockPtr_ = fileDeleteLock;
       filesToDeletePtr_ = filesToDelete;
     }

◆ setFMS()

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

inline

Definition at line 121 of file EvFDaqDirector.h.

References fms_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

121 { fms_ = fms; }

evf::EvFDaqDirector::fms_

evf::FastMonitoringService * fms_

Definition: EvFDaqDirector.h:257

◆ tryInitializeFuLockFile()

void evf::EvFDaqDirector::tryInitializeFuLockFile ( )

Definition at line 880 of file EvFDaqDirector.cc.

References fu_rw_lock_stream, and getHLTprescales::readIndex().

Referenced by initRun().

                                                {
     if (fu_rw_lock_stream == nullptr)
       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);
     }
   }

◆ unlockFULocal()

void evf::EvFDaqDirector::unlockFULocal ( )

Definition at line 918 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by getNextFromFileBroker(), grabNextJsonFileAndUnlock(), FedRawDataInputSource::readSupervisor(), and ~EvFDaqDirector().

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

◆ unlockFULocal2()

void evf::EvFDaqDirector::unlockFULocal2 ( )

Definition at line 928 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

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

◆ unlockInitLock()

void evf::EvFDaqDirector::unlockInitLock ( )

Definition at line 911 of file EvFDaqDirector.cc.

References init_lock_.

911 { pthread_mutex_unlock(&init_lock_); }

evf::EvFDaqDirector::init_lock_

pthread_mutex_t init_lock_

Definition: EvFDaqDirector.h:262

◆ updateFuLock()

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

Definition at line 503 of file EvFDaqDirector.cc.

References bu_run_dir_, visDQMUpload::buf, bumpFile(), RPCNoise_example::check, fu_readwritelock_fd_, fu_rw_flk, fu_rw_fulk, fulockfile_, fuLockPollInterval_, getEoLSFilePathOnFU(), getEoRFilePath(), lockFULocal(), LogDebug, eostools::ls(), newFile, noFile, getHLTprescales::readIndex(), readLastLSEntry(), runAbort, runEnded, edm_modernize_messagelogger::stat, stop_ls_override_, stopFilePath_, and stopFilePathPid_.

Referenced by FedRawDataInputSource::readSupervisor().

                                                                                    {
     EvFDaqDirector::FileStatus fileStatus = noFile;
     rawHeaderSize = 0;
 
     int retval = -1;
     int lock_attempts = 0;
     long total_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, nullptr);
 
     while (retval == -1) {
       retval = fcntl(fu_readwritelock_fd_, F_SETLK, &fu_rw_flk);
       if (retval == -1)
         usleep(fuLockPollInterval_);
       else
         continue;
 
       lock_attempts += fuLockPollInterval_;
       total_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(fulockfile_.c_str(), &buf) != 0)
           return runEnded;
 
         lock_attempts = 0;
       }
       if (total_lock_attempts > 5 * 60000000) {
         edm::LogError("EvFDaqDirector") << "Unable to obtain a lock for 5 minutes. Stopping polling activity.";
         return runAbort;
       }
     }
 
     timeval ts_lockend;
     gettimeofday(&ts_lockend, nullptr);
     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
 
     //open another lock file FD after the lock using main fd has been acquired
     int fu_readwritelock_fd2 = open(fulockfile_.c_str(), O_RDWR, S_IRWXU);
     if (fu_readwritelock_fd2 == -1)
       edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile_
                                       << " create. error:" << strerror(errno);
 
     FILE* fu_rw_lock_stream2 = fdopen(fu_readwritelock_fd2, "r+");
 
     // if the stream is readable
     if (fu_rw_lock_stream2 != nullptr) {
       unsigned int readLs, readIndex;
       int check = 0;
       // rewind the stream
       check = fseek(fu_rw_lock_stream2, 0, SEEK_SET);
       // if rewinded ok
       if (check == 0) {
         // read its' values
         fscanf(fu_rw_lock_stream2, "%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, rawHeaderSize, stopFileLS, setExceptionState);
           //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
               //this covers case where run has empty first lumisections and CMSSW are late to the lock file. always one process will start with LS 1,... and create empty files for them
               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_stream2, 0, SEEK_SET);
           if (check == 0) {
             ftruncate(fu_readwritelock_fd2, 0);
             // write next index in the file, which is the file the next process should take
             fprintf(fu_rw_lock_stream2, "%u %u", readLs, readIndex + 1);
             fflush(fu_rw_lock_stream2);
             fsync(fu_readwritelock_fd2);
             fileStatus = newFile;
             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);
             setExceptionState = true;
             return noFile;
           }
         } 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_stream2, 0, SEEK_SET);
           if (check == 0) {
             ftruncate(fu_readwritelock_fd2, 0);
             // in this case LS was bumped, but no new file. Thus readIndex is 0 (set by bumpFile)
             fprintf(fu_rw_lock_stream2, "%u %u", readLs, readIndex);
             fflush(fu_rw_lock_stream2);
             fsync(fu_readwritelock_fd2);
             LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":" << readIndex;
           } else {
             edm::LogError("EvFDaqDirector")
                 << "seek on fu read/write lock for updating failed with error " << strerror(errno);
             setExceptionState = true;
             return noFile;
           }
         }
       } 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);
     }
     fclose(fu_rw_lock_stream2);  // = fdopen(fu_readwritelock_fd2, "r+");
 
 #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;
   }

◆ useFileBroker()

bool evf::EvFDaqDirector::useFileBroker ( ) const

inline

Definition at line 79 of file EvFDaqDirector.h.

References useFileBroker_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

79 { return useFileBroker_; }

evf::EvFDaqDirector::useFileBroker_

bool useFileBroker_

Definition: EvFDaqDirector.h:209

Member Data Documentation

◆ base_dir_

std::string evf::EvFDaqDirector::base_dir_

private

Definition at line 206 of file EvFDaqDirector.h.

Referenced by initRun().

◆ bu_base_dir_

std::string evf::EvFDaqDirector::bu_base_dir_

private

Definition at line 207 of file EvFDaqDirector.h.

Referenced by initRun().

◆ bu_r_flk

struct flock evf::EvFDaqDirector::bu_r_flk

private

Definition at line 251 of file EvFDaqDirector.h.

◆ bu_r_fulk

struct flock evf::EvFDaqDirector::bu_r_fulk

private

Definition at line 253 of file EvFDaqDirector.h.

◆ bu_r_lock_stream

FILE* evf::EvFDaqDirector::bu_r_lock_stream

private

Definition at line 241 of file EvFDaqDirector.h.

◆ bu_readlock_fd_

int evf::EvFDaqDirector::bu_readlock_fd_

private

Definition at line 234 of file EvFDaqDirector.h.

Referenced by postEndRun().

◆ bu_run_dir_

std::string evf::EvFDaqDirector::bu_run_dir_

private

Definition at line 230 of file EvFDaqDirector.h.

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

◆ bu_run_open_dir_

std::string evf::EvFDaqDirector::bu_run_open_dir_

private

Definition at line 231 of file EvFDaqDirector.h.

Referenced by buBaseRunOpenDir(), and initRun().

◆ bu_t_monitor_stream

FILE* evf::EvFDaqDirector::bu_t_monitor_stream

private

Definition at line 244 of file EvFDaqDirector.h.

◆ bu_w_flk

struct flock evf::EvFDaqDirector::bu_w_flk

private

Definition at line 250 of file EvFDaqDirector.h.

◆ bu_w_fulk

struct flock evf::EvFDaqDirector::bu_w_fulk

private

Definition at line 252 of file EvFDaqDirector.h.

◆ bu_w_lock_stream

FILE* evf::EvFDaqDirector::bu_w_lock_stream

private

Definition at line 240 of file EvFDaqDirector.h.

Referenced by initRun().

◆ bu_w_monitor_stream

FILE* evf::EvFDaqDirector::bu_w_monitor_stream

private

Definition at line 243 of file EvFDaqDirector.h.

◆ bu_writelock_fd_

int evf::EvFDaqDirector::bu_writelock_fd_

private

Definition at line 235 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

◆ directorBU_

bool evf::EvFDaqDirector::directorBU_

private

Definition at line 220 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

◆ dirManager_

DirManager evf::EvFDaqDirector::dirManager_

private

Definition at line 246 of file EvFDaqDirector.h.

Referenced by findCurrentRunDir(), and preBeginRun().

◆ dpd_

jsoncollector::DataPointDefinition* evf::EvFDaqDirector::dpd_

private

Definition at line 280 of file EvFDaqDirector.h.

Referenced by grabNextJsonFile(), grabNextJsonFileAndUnlock(), and initRun().

◆ endpoint_iterator_

std::unique_ptr<boost::asio::ip::tcp::resolver::iterator> evf::EvFDaqDirector::endpoint_iterator_

private

Definition at line 285 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), and EvFDaqDirector().

◆ eolsNFilesIndex_

unsigned int evf::EvFDaqDirector::eolsNFilesIndex_ = 1

private

Definition at line 268 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

◆ fileBrokerHost_

std::string evf::EvFDaqDirector::fileBrokerHost_

private

Definition at line 211 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), and EvFDaqDirector().

◆ fileBrokerHostFromCfg_

bool evf::EvFDaqDirector::fileBrokerHostFromCfg_

private

Definition at line 210 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

◆ fileBrokerKeepAlive_

bool evf::EvFDaqDirector::fileBrokerKeepAlive_

private

Definition at line 213 of file EvFDaqDirector.h.

Referenced by contactFileBroker().

◆ fileBrokerPort_

std::string evf::EvFDaqDirector::fileBrokerPort_

private

Definition at line 212 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

◆ fileBrokerUseLocalLock_

bool evf::EvFDaqDirector::fileBrokerUseLocalLock_

private

Definition at line 214 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and getNextFromFileBroker().

◆ fileDeleteLockPtr_

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

private

Definition at line 259 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

◆ filesToDeletePtr_

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

private

Definition at line 260 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

◆ fms_

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

private

Definition at line 257 of file EvFDaqDirector.h.

Referenced by bumpFile(), preGlobalEndLumi(), and setFMS().

◆ fu_readwritelock_fd_

int evf::EvFDaqDirector::fu_readwritelock_fd_

private

Definition at line 236 of file EvFDaqDirector.h.

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

◆ fu_rw_flk

struct flock evf::EvFDaqDirector::fu_rw_flk

private

Definition at line 254 of file EvFDaqDirector.h.

Referenced by updateFuLock().

◆ fu_rw_fulk

struct flock evf::EvFDaqDirector::fu_rw_fulk

private

Definition at line 255 of file EvFDaqDirector.h.

Referenced by updateFuLock().

◆ fu_rw_lock_stream

FILE* evf::EvFDaqDirector::fu_rw_lock_stream

private

Definition at line 242 of file EvFDaqDirector.h.

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

◆ fulocal_rwlock_fd2_

int evf::EvFDaqDirector::fulocal_rwlock_fd2_

private

Definition at line 238 of file EvFDaqDirector.h.

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

◆ fulocal_rwlock_fd_

int evf::EvFDaqDirector::fulocal_rwlock_fd_

private

Definition at line 237 of file EvFDaqDirector.h.

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

◆ fulockfile_

std::string evf::EvFDaqDirector::fulockfile_

private

Definition at line 232 of file EvFDaqDirector.h.

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

◆ fuLockPollInterval_

unsigned int evf::EvFDaqDirector::fuLockPollInterval_

private

Definition at line 215 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateFuLock().

◆ hltSourceDirectory_

std::string evf::EvFDaqDirector::hltSourceDirectory_

private

Definition at line 221 of file EvFDaqDirector.h.

Referenced by initRun().

◆ hostname_

std::string evf::EvFDaqDirector::hostname_

private

Definition at line 225 of file EvFDaqDirector.h.

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

◆ init_lock_

pthread_mutex_t evf::EvFDaqDirector::init_lock_ = PTHREAD_MUTEX_INITIALIZER

private

Definition at line 262 of file EvFDaqDirector.h.

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

◆ input_throttled_file_

std::string evf::EvFDaqDirector::input_throttled_file_

private

Definition at line 288 of file EvFDaqDirector.h.

Referenced by initRun(), and inputThrottled().

◆ io_service_

boost::asio::io_service evf::EvFDaqDirector::io_service_

private

Definition at line 282 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

◆ mergeTypeMap_

tbb::concurrent_hash_map<std::string, std::string> evf::EvFDaqDirector::mergeTypeMap_

private

Definition at line 274 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet(), and getStreamMergeType().

◆ MergeTypeNames_

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

staticprivate

Definition at line 277 of file EvFDaqDirector.h.

Referenced by getStreamMergeType().

◆ mergeTypePset_

std::string evf::EvFDaqDirector::mergeTypePset_

private

Definition at line 219 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet().

◆ nStreams_

unsigned int evf::EvFDaqDirector::nStreams_ = 0

private

Definition at line 264 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

◆ nThreads_

unsigned int evf::EvFDaqDirector::nThreads_ = 0

private

Definition at line 265 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

◆ outputAdler32Recheck_

bool evf::EvFDaqDirector::outputAdler32Recheck_

private

Definition at line 216 of file EvFDaqDirector.h.

Referenced by outputAdler32Recheck().

◆ pid_

std::string evf::EvFDaqDirector::pid_

private

Definition at line 228 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), grabNextJsonFile(), grabNextJsonFromRaw(), and initRun().

◆ previousFileSize_

unsigned long evf::EvFDaqDirector::previousFileSize_

private

Definition at line 248 of file EvFDaqDirector.h.

Referenced by bumpFile().

◆ query_

std::unique_ptr<boost::asio::ip::tcp::resolver::query> evf::EvFDaqDirector::query_

private

Definition at line 284 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

◆ readEolsDefinition_

bool evf::EvFDaqDirector::readEolsDefinition_ = true

private

Definition at line 267 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

◆ requireTSPSet_

bool evf::EvFDaqDirector::requireTSPSet_

private

Definition at line 217 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().

◆ resolver_

std::unique_ptr<boost::asio::ip::tcp::resolver> evf::EvFDaqDirector::resolver_

private

Definition at line 283 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

◆ run_

unsigned int evf::EvFDaqDirector::run_

private

◆ run_dir_

std::string evf::EvFDaqDirector::run_dir_

private

◆ run_nstring_

std::string evf::EvFDaqDirector::run_nstring_

private

Definition at line 227 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), and initRun().

◆ run_string_

std::string evf::EvFDaqDirector::run_string_

private

Definition at line 226 of file EvFDaqDirector.h.

Referenced by getLumisectionToStart(), and initRun().

◆ selectedTransferMode_

std::string evf::EvFDaqDirector::selectedTransferMode_

private

Definition at line 218 of file EvFDaqDirector.h.

Referenced by getStreamDestinations().

◆ socket_

std::unique_ptr<boost::asio::ip::tcp::socket> evf::EvFDaqDirector::socket_

private

Definition at line 286 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), EvFDaqDirector(), and ~EvFDaqDirector().

◆ startFromLS_

unsigned int evf::EvFDaqDirector::startFromLS_ = 1

private

Definition at line 223 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and getStartLumisectionFromEnv().

◆ stop_ls_override_

unsigned int evf::EvFDaqDirector::stop_ls_override_ = 0

private

Definition at line 271 of file EvFDaqDirector.h.

Referenced by getNextFromFileBroker(), and updateFuLock().

◆ stopFilePath_

std::string evf::EvFDaqDirector::stopFilePath_

private

Definition at line 269 of file EvFDaqDirector.h.

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

◆ stopFilePathPid_

std::string evf::EvFDaqDirector::stopFilePathPid_

private

Definition at line 270 of file EvFDaqDirector.h.

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

◆ transferSystemJson_

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

private

Definition at line 273 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().

◆ useFileBroker_

bool evf::EvFDaqDirector::useFileBroker_

private

Definition at line 209 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and useFileBroker().

Public Types

Public Member Functions

Static Public Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Member Enumeration Documentation

◆ FileStatus

Constructor & Destructor Documentation

◆ EvFDaqDirector()

◆ ~EvFDaqDirector()

Member Function Documentation

◆ baseRunDir()

◆ buBaseRunDir()

◆ buBaseRunOpenDir()

◆ bumpFile()

◆ checkMergeTypePSet()

◆ checkTransferSystemPSet()

◆ contactFileBroker()

◆ createBoLSFile()

◆ createLumiSectionFiles()

◆ createProcessingNotificationMaybe()

◆ createRunOpendirMaybe()

◆ eolsFileName()

◆ eorFileName()

◆ fillDescriptions()

◆ findCurrentRunDir()

◆ getBoLSFilePathOnFU()

◆ getDatFilePath()

◆ getEoLSFilePathOnBU()

◆ getEoLSFilePathOnFU()

◆ getEoRFilePath()

◆ getEoRFilePathOnFU()

◆ getFFFParamsFilePathOnBU()

◆ getInitFilePath()

◆ getInputJsonFilePath()

◆ getLumisectionToStart()

◆ getMergedDatChecksumFilePath()

◆ getMergedDatFilePath()

◆ getMergedProtocolBufferHistogramFilePath()

◆ getMergedRootHistogramFilePath()

◆ getNextFromFileBroker()

◆ getNFilesFromEoLS()

◆ getOpenDatFilePath()

◆ getOpenInitFilePath()

◆ getOpenInputJsonFilePath()

◆ getOpenOutputJsonFilePath()

◆ getOpenProtocolBufferHistogramFilePath()

◆ getOpenRawFilePath()

◆ getOpenRootHistogramFilePath()

◆ getOutputJsonFilePath()

◆ getProtocolBufferHistogramFilePath()

◆ getRawFilePath()

◆ getRootHistogramFilePath()

◆ getRunNumber()

◆ getRunOpenDirPath()

◆ getStartLumisectionFromEnv()

◆ getStreamDestinations()

◆ getStreamMergeType()

◆ grabNextJsonFile()

◆ grabNextJsonFileAndUnlock()

◆ grabNextJsonFromRaw()

◆ initFileName()

◆ initRun()

◆ inputFileNameStem()

◆ inputThrottled()

◆ isSingleStreamThread()

◆ lockFULocal()

◆ lockFULocal2()

◆ lockInitLock()

◆ make_flock()

◆ mergedFileNameStem()

◆ openFULockfileStream()

◆ outputAdler32Recheck()

◆ outputFileNameStem()

◆ overrideRunNumber()

◆ parseFRDFileHeader()

◆ postEndRun()

◆ preallocate()