evf::EvFDaqDirector Class Reference

#include <EvFDaqDirector.h>

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

Public Member Functions
std::string &	baseRunDir ()
std::string &	buBaseRunDir ()
std::string &	buBaseRunOpenDir ()
void	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::vector< std::string > const &	getBUBaseDirs () 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	getEoRFileName () const
std::string	getEoRFilePath () const
std::string	getEoRFilePathOnFU () const
std::string	getFFFParamsFilePathOnBU () const
std::string	getInitFilePath (std::string const &stream) const
std::string	getInitTempFilePath (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, bool requireHeader=true)
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, bool requireHeader=true)
void	initRun ()
bool	inputThrottled ()
bool	isSingleStreamThread ()
void	lockFULocal ()
void	lockFULocal2 ()
void	lockInitLock ()
bool	lumisectionDiscarded (unsigned int ls)
unsigned int	numConcurrentLumis () const
bool	outputAdler32Recheck () const
void	overrideRunNumber (unsigned int run)
void	postEndRun (edm::GlobalContext const &globalContext)
void	preallocate (edm::service::SystemBounds const &bounds)
void	preBeginJob (edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &)
void	preBeginRun (edm::GlobalContext const &globalContext)
void	preGlobalEndLumi (edm::GlobalContext const &globalContext)
bool	rawFileHasHeader (std::string const &rawSourcePath, uint16_t &rawHeaderSize)
int	readLastLSEntry (std::string const &file)
void	removeFile (std::string)
std::string const &	runString () const
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, uint16_t &rawDataType, 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

Static Private Member Functions
static bool	checkFileRead (char *buf, int infile, std::size_t buf_sz, std::string const &path)

Private Attributes
std::string	base_dir_
std::string	bu_base_dir_
std::vector< std::string >	bu_base_dirs_all_
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_
std::string	discard_ls_filestem_
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	nConcurrentLumis_ = 0
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.

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

Constructor & Destructor Documentation

EvFDaqDirector()

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

explicit

Definition at line 39 of file EvFDaqDirector.cc.

References base_dir_, visDQMUpload::buf, discard_ls_filestem_, endpoint_iterator_, cppFunctionSkipper::exception, fileBrokerHost_, fileBrokerHostFromCfg_, fileBrokerPort_, fileBrokerUseLocalLock_, fuLockPollInterval_, hostname_, recoMuon::in, input_throttled_file_, io_service_, pid_, postEndRun(), preallocate(), preBeginJob(), preBeginRun(), preGlobalEndLumi(), query_, resolver_, run_, run_dir_, run_nstring_, run_string_, socket_, contentValuesCheck::ss, 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")),
        bu_base_dirs_all_(pset.getUntrackedParameter<std::vector<std::string>>("buBaseDirsAll")),
        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);
      }
    }

    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";
    discard_ls_filestem_ = run_dir_ + "/discard_ls";
    ss = std::stringstream();
    ss << getpid();
    pid_ = ss.str();
  }

~EvFDaqDirector()

evf::EvFDaqDirector::~EvFDaqDirector

(

)

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

References run_dir_.

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

{ return run_dir_; }

buBaseRunDir()

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

inline

Definition at line 78 of file EvFDaqDirector.h.

References bu_run_dir_.

Referenced by DAQSource::readSupervisor().

{ return bu_run_dir_; }

buBaseRunOpenDir()

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

inline

Definition at line 79 of file EvFDaqDirector.h.

References bu_run_open_dir_.

{ return bu_run_open_dir_; }

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 818 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;
  }

checkFileRead()

bool evf::EvFDaqDirector::checkFileRead ( char *  buf, int  infile, std::size_t  buf_sz, std::string const &  path  )

staticprivate

Definition at line 1112 of file EvFDaqDirector.cc.

References visDQMUpload::buf, timingPdfMaker::infile, castor_dqm_sourceclient_file_cfg::path, and fileinputsource_cfi::read.

Referenced by parseFRDFileHeader(), and rawFileHasHeader().

                                                                                                   {
    ssize_t sz_read = ::read(infile, buf, buf_sz);
    if (sz_read < 0) {
      edm::LogError("EvFDaqDirector") << "rawFileHasHeader - unable to read " << path << " : " << strerror(errno);
      if (infile != -1)
        close(infile);
      return false;
    }
    if ((size_t)sz_read < buf_sz) {
      edm::LogError("EvFDaqDirector") << "rawFileHasHeader - file smaller than header: " << path;
      if (infile != -1)
        close(infile);
      return false;
    }
    return true;
  }

checkMergeTypePSet()

void evf::EvFDaqDirector::checkMergeTypePSet

(

edm::ProcessContext const &

pc

)

Definition at line 2095 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 1987 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 1532 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 972 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 982 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 2127 of file EvFDaqDirector.cc.

References run_dir_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::checkNext(), and DAQSource::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 1948 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 360 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<std::vector<std::string>>("buBaseDirsAll", std::vector<std::string>())
        ->setComment("BU base ramdisk directories for multi-file DAQSource models");
    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 82 of file EvFDaqDirector.h.

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

{ return dirManager_.findRunDir(run_); }

getBoLSFilePathOnFU()

std::string evf::EvFDaqDirector::getBoLSFilePathOnFU

(

const unsigned int

ls

)

const

Definition at line 523 of file EvFDaqDirector.cc.

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

Referenced by createBoLSFile().

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

getBUBaseDirs()

std::vector<std::string> const& evf::EvFDaqDirector::getBUBaseDirs ( ) const

inline

Definition at line 196 of file EvFDaqDirector.h.

References bu_base_dirs_all_.

Referenced by DAQSource::DAQSource().

{ return bu_base_dirs_all_; }

getDatFilePath()

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

Definition at line 452 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 515 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 519 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);
  }

getEoRFileName()

std::string evf::EvFDaqDirector::getEoRFileName

(

)

const

Definition at line 529 of file EvFDaqDirector.cc.

References fffnaming::eorFileName(), and run_.

Referenced by DAQSource::readSupervisor().

{ return fffnaming::eorFileName(run_); }

getEoRFilePath()

std::string evf::EvFDaqDirector::getEoRFilePath

(

)

const

Definition at line 527 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

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

getEoRFilePathOnFU()

std::string evf::EvFDaqDirector::getEoRFilePathOnFU

(

)

const

Definition at line 531 of file EvFDaqDirector.cc.

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

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

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

getFFFParamsFilePathOnBU()

std::string evf::EvFDaqDirector::getFFFParamsFilePathOnBU

(

)

const

Definition at line 533 of file EvFDaqDirector.cc.

References bu_run_dir_.

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

getInitFilePath()

std::string evf::EvFDaqDirector::getInitFilePath

(

std::string const &

stream

)

const

Definition at line 480 of file EvFDaqDirector.cc.

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

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

getInitTempFilePath()

std::string evf::EvFDaqDirector::getInitTempFilePath

(

std::string const &

stream

)

const

Definition at line 484 of file EvFDaqDirector.cc.

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

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

getInputJsonFilePath()

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

Definition at line 436 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 1972 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(), and DAQSource::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 472 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 468 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 498 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 511 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,
		bool	requireHeader = true
	)

Definition at line 1798 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(), and DAQSource::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, requireHeader);
      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 757 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 456 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 476 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 448 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 460 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 488 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 444 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 503 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 464 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 493 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 440 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

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

References run_.

{ return run_; }

getRunOpenDirPath()

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

inline

Definition at line 109 of file EvFDaqDirector.h.

References run_dir_.

Referenced by createRunOpendirMaybe(), and initRun().

{ return run_dir_ + "/open"; }

getStartLumisectionFromEnv()

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

inline

Definition at line 182 of file EvFDaqDirector.h.

References startFromLS_.

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

{ return startFromLS_; }

getStreamDestinations()

std::string evf::EvFDaqDirector::getStreamDestinations

(

std::string const &

stream

)

const

Definition at line 2049 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 2113 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 1257 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 1440 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,
		bool	requireHeader = true
	)

Definition at line 1165 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;
    uint16_t rawDataType;
    auto ret = parseFRDFileHeader(rawSourcePath,
                                  rawFd,
                                  rawHeaderSize,
                                  rawDataType,
                                  lsFromRaw,
                                  nbEventsWrittenRaw,
                                  fileSizeFromRaw,
                                  requireHeader,
                                  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 163 of file EvFDaqDirector.cc.

References base_dir_, bu_base_dir_, bu_base_dirs_all_, 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_, mps_fire::i, init_lock_, svgfig::load(), eostools::mkdir(), openFULockfileStream(), pid_, run_dir_, run_string_, edm::shutdown_flag, edm_modernize_messagelogger::stat, stopFilePath_, stopFilePathPid_, AlCaHLTBitMon_QueryRunRegistry::string, tryInitializeFuLockFile(), and useFileBroker_.

Referenced by preallocate().

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

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

          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

      auto checkExists = [=](std::string const& bu_base_dir) -> void {
        int 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);
        }
      };

      auto waitForDir = [=](std::string const& bu_base_dir) -> void {
        int cnt = 0;
        while (!edm::shutdown_flag.load(std::memory_order_relaxed)) {
          int retval = mkdir(bu_base_dir.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
          if (retval != 0 && errno != EEXIST) {
            usleep(500000);
            cnt++;
            if (cnt % 20 == 0)
              edm::LogWarning("DaqDirector") << "waiting for " << bu_base_dir;
            if (cnt > 120)
              throw cms::Exception("DaqDirector") << " Error checking for bu base dir after 1 minute -: " << bu_base_dir
                                                  << " mkdir error:" << strerror(errno);
          }
          break;
        }
      };

      if (!bu_base_dirs_all_.empty()) {
        checkExists(bu_base_dirs_all_[0]);
        bu_run_dir_ = bu_base_dirs_all_[0] + "/" + run_string_;
        for (unsigned int i = 1; i < bu_base_dirs_all_.size(); i++)
          waitForDir(bu_base_dirs_all_[i]);
      } else {
        checkExists(bu_base_dir_);
        bu_run_dir_ = bu_base_dir_ + "/" + run_string_;
      }

      fulockfile_ = bu_run_dir_ + "/fu.lock";
      if (!useFileBroker_)
        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 2141 of file EvFDaqDirector.cc.

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

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

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

isSingleStreamThread()

bool evf::EvFDaqDirector::isSingleStreamThread ( )

inline

Definition at line 124 of file EvFDaqDirector.h.

References nStreams_, and nThreads_.

Referenced by DAQSource::getNextDataBlock(), and FedRawDataInputSource::getNextEvent().

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

lockFULocal()

void evf::EvFDaqDirector::lockFULocal

(

)

Definition at line 952 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 962 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 948 of file EvFDaqDirector.cc.

References init_lock_.

{ pthread_mutex_lock(&init_lock_); }

lumisectionDiscarded()

bool evf::EvFDaqDirector::lumisectionDiscarded

(

unsigned int

ls

)

Definition at line 2146 of file EvFDaqDirector.cc.

References visDQMUpload::buf, discard_ls_filestem_, eostools::ls(), edm_modernize_messagelogger::stat, and to_string().

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

                                                           {
    struct stat buf;
    return (stat((discard_ls_filestem_ + std::to_string(ls)).c_str(), &buf) == 0);
  }

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 2133 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

numConcurrentLumis()

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

inline

Definition at line 125 of file EvFDaqDirector.h.

References nConcurrentLumis_.

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

{ return nConcurrentLumis_; }

openFULockfileStream()

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

private

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

References outputAdler32Recheck_.

{ return outputAdler32Recheck_; }

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

Referenced by DAQSource::DAQSource().

{ run_ = run; }

parseFRDFileHeader()

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

static

Definition at line 1004 of file EvFDaqDirector.cc.

References checkFileRead(), FRDFileHeaderContent_v2::dataType_, FRDFileHeaderContent_v1::eventCount_, FRDFileHeaderContent_v2::eventCount_, FRDFileHeaderContent_v1::fileSize_, FRDFileHeaderContent_v2::fileSize_, getFRDFileHeaderVersion(), FRDFileHeaderContent_v1::headerSize_, FRDFileHeaderContent_v2::headerSize_, FRDFileHeaderIdentifier::id_, timingPdfMaker::infile, FRDFileHeaderContent_v1::lumiSection_, FRDFileHeaderContent_v2::lumiSection_, and FRDFileHeaderIdentifier::version_.

Referenced by grabNextJsonFromRaw(), FedRawDataInputSource::readSupervisor(), and DAQSource::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,
                                  rawDataType,
                                  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;
        }
      }
    }

    //v2 is the largest possible read
    char hdr[sizeof(FRDFileHeader_v2)];
    if (!checkFileRead(hdr, infile, sizeof(FRDFileHeaderIdentifier), rawSourcePath))
      return -1;

    FRDFileHeaderIdentifier* fileId = (FRDFileHeaderIdentifier*)hdr;
    uint16_t frd_version = getFRDFileHeaderVersion(fileId->id_, fileId->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;
        close(infile);
        return -1;
      } else {
        //no header, but valid file
        lseek(infile, 0, SEEK_SET);
        rawHeaderSize = 0;
        lsFromHeader = 0;
        eventsFromHeader = -1;
        fileSizeFromHeader = -1;
      }
    } else if (frd_version == 1) {
      //version 1 header
      if (!checkFileRead(hdr, infile, sizeof(FRDFileHeaderContent_v1), rawSourcePath))
        return -1;
      FRDFileHeaderContent_v1* fhContent = (FRDFileHeaderContent_v1*)hdr;
      uint32_t headerSizeRaw = fhContent->headerSize_;
      if (headerSizeRaw != sizeof(FRDFileHeader_v1)) {
        edm::LogError("EvFDaqDirector") << "inconsistent header size: " << rawSourcePath << " size: " << headerSizeRaw
                                        << " v:" << frd_version;
        close(infile);
        return -1;
      }
      //allow header size to exceed read size. Future header versions will not break this, but the size can change.
      rawDataType = 0;
      lsFromHeader = fhContent->lumiSection_;
      eventsFromHeader = (int32_t)fhContent->eventCount_;
      fileSizeFromHeader = (int64_t)fhContent->fileSize_;
      rawHeaderSize = fhContent->headerSize_;

    } else if (frd_version == 2) {
      //version 2 heade
      if (!checkFileRead(hdr, infile, sizeof(FRDFileHeaderContent_v2), rawSourcePath))
        return -1;
      FRDFileHeaderContent_v2* fhContent = (FRDFileHeaderContent_v2*)hdr;
      uint32_t headerSizeRaw = fhContent->headerSize_;
      if (headerSizeRaw != sizeof(FRDFileHeader_v2)) {
        edm::LogError("EvFDaqDirector") << "inconsistent header size: " << rawSourcePath << " size: " << headerSizeRaw
                                        << " v:" << frd_version;
        close(infile);
        return -1;
      }
      //allow header size to exceed read size. Future header versions will not break this, but the size can change.
      rawDataType = fhContent->dataType_;
      lsFromHeader = fhContent->lumiSection_;
      eventsFromHeader = (int32_t)fhContent->eventCount_;
      fileSizeFromHeader = (int64_t)fhContent->fileSize_;
      rawHeaderSize = fhContent->headerSize_;
    }

    if (closeFile) {
      close(infile);
      infile = -1;
    }

    rawFd = infile;
    return 0;  //OK
  }

postEndRun()

void evf::EvFDaqDirector::postEndRun

(

edm::GlobalContext const &

globalContext

)

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

References initRun(), nConcurrentLumis_, nStreams_, and nThreads_.

Referenced by EvFDaqDirector().

                                                                       {
    initRun();

    nThreads_ = bounds.maxNumberOfStreams();
    nStreams_ = bounds.maxNumberOfThreads();
    nConcurrentLumis_ = bounds.maxNumberOfConcurrentLuminosityBlocks();
  }

preBeginJob()

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

Definition at line 394 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 399 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 418 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 1129 of file EvFDaqDirector.cc.

References checkFileRead(), getFRDFileHeaderVersion(), FRDFileHeaderContent_v1::headerSize_, FRDFileHeaderContent_v2::headerSize_, FRDFileHeaderIdentifier::id_, timingPdfMaker::infile, and FRDFileHeaderIdentifier::version_.

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;
    }
    //try to read FRD header size (v2 is the biggest, use read buffer of that size)
    char hdr[sizeof(FRDFileHeader_v2)];
    if (!checkFileRead(hdr, infile, sizeof(FRDFileHeaderIdentifier), rawSourcePath))
      return false;
    FRDFileHeaderIdentifier* fileId = (FRDFileHeaderIdentifier*)hdr;
    uint16_t frd_version = getFRDFileHeaderVersion(fileId->id_, fileId->version_);

    if (frd_version == 1) {
      if (!checkFileRead(hdr, infile, sizeof(FRDFileHeaderContent_v1), rawSourcePath))
        return false;
      FRDFileHeaderContent_v1* fhContent = (FRDFileHeaderContent_v1*)hdr;
      rawHeaderSize = fhContent->headerSize_;
      close(infile);
      return true;
    } else if (frd_version == 2) {
      if (!checkFileRead(hdr, infile, sizeof(FRDFileHeaderContent_v2), rawSourcePath))
        return false;
      FRDFileHeaderContent_v2* fhContent = (FRDFileHeaderContent_v2*)hdr;
      rawHeaderSize = fhContent->headerSize_;
      close(infile);
      return true;
    } else
      edm::LogError("EvFDaqDirector") << "rawFileHasHeader - unknown version: " << frd_version;

    close(infile);
    rawHeaderSize = 0;
    return false;
  }

readLastLSEntry()

int evf::EvFDaqDirector::readLastLSEntry

(

std::string const &

file

)

Definition at line 1958 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()

void evf::EvFDaqDirector::removeFile

(

std::string

filename

)

Definition at line 535 of file EvFDaqDirector.cc.

References corrVsCorr::filename.

Referenced by postEndRun().

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

runString()

std::string const& evf::EvFDaqDirector::runString ( ) const

inline

Definition at line 76 of file EvFDaqDirector.h.

References run_string_.

Referenced by DAQSource::DAQSource().

{ return run_string_; }

setDeleteTracking()

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

inline

Definition at line 183 of file EvFDaqDirector.h.

References fileDeleteLockPtr_, and filesToDeletePtr_.

Referenced by DAQSource::DAQSource(), and FedRawDataInputSource::FedRawDataInputSource().

                                                                                           {
      fileDeleteLockPtr_ = fileDeleteLock;
      filesToDeletePtr_ = filesToDelete;
    }

setFMS()

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

inline

Definition at line 123 of file EvFDaqDirector.h.

References fms_.

Referenced by DAQSource::DAQSource(), and FedRawDataInputSource::FedRawDataInputSource().

{ fms_ = fms; }

tryInitializeFuLockFile()

void evf::EvFDaqDirector::tryInitializeFuLockFile

(

)

Definition at line 919 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 957 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 967 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 950 of file EvFDaqDirector.cc.

References init_lock_.

{ pthread_mutex_unlock(&init_lock_); }

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 542 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 80 of file EvFDaqDirector.h.

References useFileBroker_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

{ return useFileBroker_; }

Member Data Documentation

base_dir_

std::string evf::EvFDaqDirector::base_dir_

private

Definition at line 216 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and initRun().

bu_base_dir_

std::string evf::EvFDaqDirector::bu_base_dir_

private

Definition at line 217 of file EvFDaqDirector.h.

Referenced by initRun().

bu_base_dirs_all_

std::vector<std::string> evf::EvFDaqDirector::bu_base_dirs_all_

private

Definition at line 218 of file EvFDaqDirector.h.

Referenced by getBUBaseDirs(), and initRun().

bu_r_flk

struct flock evf::EvFDaqDirector::bu_r_flk

private

Definition at line 262 of file EvFDaqDirector.h.

bu_r_fulk

struct flock evf::EvFDaqDirector::bu_r_fulk

private

Definition at line 264 of file EvFDaqDirector.h.

bu_r_lock_stream

FILE* evf::EvFDaqDirector::bu_r_lock_stream

private

Definition at line 252 of file EvFDaqDirector.h.

bu_readlock_fd_

int evf::EvFDaqDirector::bu_readlock_fd_

private

Definition at line 245 of file EvFDaqDirector.h.

Referenced by postEndRun().

bu_run_dir_

std::string evf::EvFDaqDirector::bu_run_dir_

private

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

Referenced by buBaseRunOpenDir(), and initRun().

bu_t_monitor_stream

FILE* evf::EvFDaqDirector::bu_t_monitor_stream

private

Definition at line 255 of file EvFDaqDirector.h.

bu_w_flk

struct flock evf::EvFDaqDirector::bu_w_flk

private

Definition at line 261 of file EvFDaqDirector.h.

bu_w_fulk

struct flock evf::EvFDaqDirector::bu_w_fulk

private

Definition at line 263 of file EvFDaqDirector.h.

bu_w_lock_stream

FILE* evf::EvFDaqDirector::bu_w_lock_stream

private

Definition at line 251 of file EvFDaqDirector.h.

Referenced by initRun().

bu_w_monitor_stream

FILE* evf::EvFDaqDirector::bu_w_monitor_stream

private

Definition at line 254 of file EvFDaqDirector.h.

bu_writelock_fd_

int evf::EvFDaqDirector::bu_writelock_fd_

private

Definition at line 246 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

directorBU_

bool evf::EvFDaqDirector::directorBU_

private

Definition at line 231 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

dirManager_

DirManager evf::EvFDaqDirector::dirManager_

private

Definition at line 257 of file EvFDaqDirector.h.

Referenced by findCurrentRunDir(), and preBeginRun().

discard_ls_filestem_

std::string evf::EvFDaqDirector::discard_ls_filestem_

private

Definition at line 301 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and lumisectionDiscarded().

dpd_

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

private

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

Referenced by contactFileBroker(), and EvFDaqDirector().

eolsNFilesIndex_

unsigned int evf::EvFDaqDirector::eolsNFilesIndex_ = 1

private

Definition at line 280 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

fileBrokerHost_

std::string evf::EvFDaqDirector::fileBrokerHost_

private

Definition at line 222 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), and EvFDaqDirector().

fileBrokerHostFromCfg_

bool evf::EvFDaqDirector::fileBrokerHostFromCfg_

private

Definition at line 221 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

fileBrokerKeepAlive_

bool evf::EvFDaqDirector::fileBrokerKeepAlive_

private

Definition at line 224 of file EvFDaqDirector.h.

Referenced by contactFileBroker().

fileBrokerPort_

std::string evf::EvFDaqDirector::fileBrokerPort_

private

Definition at line 223 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

fileBrokerUseLocalLock_

bool evf::EvFDaqDirector::fileBrokerUseLocalLock_

private

Definition at line 225 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and getNextFromFileBroker().

fileDeleteLockPtr_

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

private

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

Referenced by preGlobalEndLumi(), and setDeleteTracking().

fms_

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

private

Definition at line 268 of file EvFDaqDirector.h.

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

fu_readwritelock_fd_

int evf::EvFDaqDirector::fu_readwritelock_fd_

private

Definition at line 247 of file EvFDaqDirector.h.

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

fu_rw_flk

struct flock evf::EvFDaqDirector::fu_rw_flk

private

Definition at line 265 of file EvFDaqDirector.h.

Referenced by updateFuLock().

fu_rw_fulk

struct flock evf::EvFDaqDirector::fu_rw_fulk

private

Definition at line 266 of file EvFDaqDirector.h.

Referenced by updateFuLock().

fu_rw_lock_stream

FILE* evf::EvFDaqDirector::fu_rw_lock_stream

private

Definition at line 253 of file EvFDaqDirector.h.

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

fulocal_rwlock_fd2_

int evf::EvFDaqDirector::fulocal_rwlock_fd2_

private

Definition at line 249 of file EvFDaqDirector.h.

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

fulocal_rwlock_fd_

int evf::EvFDaqDirector::fulocal_rwlock_fd_

private

Definition at line 248 of file EvFDaqDirector.h.

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

fulockfile_

std::string evf::EvFDaqDirector::fulockfile_

private

Definition at line 243 of file EvFDaqDirector.h.

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

fuLockPollInterval_

unsigned int evf::EvFDaqDirector::fuLockPollInterval_

private

Definition at line 226 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateFuLock().

hltSourceDirectory_

std::string evf::EvFDaqDirector::hltSourceDirectory_

private

Definition at line 232 of file EvFDaqDirector.h.

Referenced by initRun().

hostname_

std::string evf::EvFDaqDirector::hostname_

private

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

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

input_throttled_file_

std::string evf::EvFDaqDirector::input_throttled_file_

private

Definition at line 300 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and inputThrottled().

io_service_

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

private

Definition at line 294 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

mergeTypeMap_

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

private

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

Referenced by getStreamMergeType().

mergeTypePset_

std::string evf::EvFDaqDirector::mergeTypePset_

private

Definition at line 230 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet().

nConcurrentLumis_

unsigned int evf::EvFDaqDirector::nConcurrentLumis_ = 0

private

Definition at line 277 of file EvFDaqDirector.h.

Referenced by numConcurrentLumis(), and preallocate().

nStreams_

unsigned int evf::EvFDaqDirector::nStreams_ = 0

private

Definition at line 275 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

nThreads_

unsigned int evf::EvFDaqDirector::nThreads_ = 0

private

Definition at line 276 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

outputAdler32Recheck_

bool evf::EvFDaqDirector::outputAdler32Recheck_

private

Definition at line 227 of file EvFDaqDirector.h.

Referenced by outputAdler32Recheck().

pid_

std::string evf::EvFDaqDirector::pid_

private

Definition at line 239 of file EvFDaqDirector.h.

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

previousFileSize_

unsigned long evf::EvFDaqDirector::previousFileSize_

private

Definition at line 259 of file EvFDaqDirector.h.

Referenced by bumpFile().

query_

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

private

Definition at line 296 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

readEolsDefinition_

bool evf::EvFDaqDirector::readEolsDefinition_ = true

private

Definition at line 279 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

requireTSPSet_

bool evf::EvFDaqDirector::requireTSPSet_

private

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

Referenced by EvFDaqDirector().

run_

unsigned int evf::EvFDaqDirector::run_

private

Definition at line 219 of file EvFDaqDirector.h.

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

run_dir_

std::string evf::EvFDaqDirector::run_dir_

private

Definition at line 240 of file EvFDaqDirector.h.

Referenced by baseRunDir(), createProcessingNotificationMaybe(), EvFDaqDirector(), getBoLSFilePathOnFU(), getDatFilePath(), getEoLSFilePathOnFU(), getEoRFilePathOnFU(), getInitFilePath(), getInitTempFilePath(), getLumisectionToStart(), getMergedDatChecksumFilePath(), getMergedDatFilePath(), getMergedProtocolBufferHistogramFilePath(), getMergedRootHistogramFilePath(), getOpenDatFilePath(), getOpenInitFilePath(), getOpenOutputJsonFilePath(), getOpenProtocolBufferHistogramFilePath(), getOpenRootHistogramFilePath(), getOutputJsonFilePath(), getProtocolBufferHistogramFilePath(), getRootHistogramFilePath(), getRunOpenDirPath(), initRun(), and preBeginRun().

run_nstring_

std::string evf::EvFDaqDirector::run_nstring_

private

Definition at line 238 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), and EvFDaqDirector().

run_string_

std::string evf::EvFDaqDirector::run_string_

private

Definition at line 237 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), getLumisectionToStart(), initRun(), and runString().

selectedTransferMode_

std::string evf::EvFDaqDirector::selectedTransferMode_

private

Definition at line 229 of file EvFDaqDirector.h.

Referenced by getStreamDestinations().

socket_

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

private

Definition at line 298 of file EvFDaqDirector.h.

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

startFromLS_

unsigned int evf::EvFDaqDirector::startFromLS_ = 1

private

Definition at line 234 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and getStartLumisectionFromEnv().

stop_ls_override_

unsigned int evf::EvFDaqDirector::stop_ls_override_ = 0

private

Definition at line 283 of file EvFDaqDirector.h.

Referenced by getNextFromFileBroker(), and updateFuLock().

stopFilePath_

std::string evf::EvFDaqDirector::stopFilePath_

private

Definition at line 281 of file EvFDaqDirector.h.

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

stopFilePathPid_

std::string evf::EvFDaqDirector::stopFilePathPid_

private

Definition at line 282 of file EvFDaqDirector.h.

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

transferSystemJson_

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

private

Definition at line 285 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().

useFileBroker_

bool evf::EvFDaqDirector::useFileBroker_

private

Definition at line 220 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), initRun(), and useFileBroker().