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, int maxLS)
void	createBoLSFile (const uint32_t lumiSection, bool checkIfExists) const
void	createLumiSectionFiles (const uint32_t lumiSection, const uint32_t currentLumiSection, bool doCreateBoLS=true)
void	createProcessingNotificationMaybe () const
void	createRunOpendirMaybe ()
	EvFDaqDirector (const edm::ParameterSet &pset, edm::ActivityRegistry &reg)
std::string	findCurrentRunDir ()
std::string	getBoLSFilePathOnFU (const unsigned int ls) const
std::string	getDatFilePath (const unsigned int ls, std::string const &stream) const
std::string	getEoLSFilePathOnBU (const unsigned int ls) const
std::string	getEoLSFilePathOnFU (const unsigned int ls) const
std::string	getEoRFilePath () const
std::string	getEoRFilePathOnFU () const
std::string	getFFFParamsFilePathOnBU () const
std::string	getInitFilePath (std::string const &stream) const
std::string	getInputJsonFilePath (const unsigned int ls, const unsigned int index) const
unsigned int	getLumisectionToStart () const
std::string	getMergedDatChecksumFilePath (const unsigned int ls, std::string const &stream) const
std::string	getMergedDatFilePath (const unsigned int ls, std::string const &stream) const
std::string	getMergedProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getMergedRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
FileStatus	getNextFromFileBroker (const unsigned int currentLumiSection, unsigned int &ls, std::string &nextFile, int &serverEventsInNewFile_, int64_t &fileSize, uint64_t &thisLockWaitTimeUs)
std::string	getOpenDatFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenInitFilePath (std::string const &stream) const
std::string	getOpenInputJsonFilePath (const unsigned int ls, const unsigned int index) const
std::string	getOpenOutputJsonFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOpenRawFilePath (const unsigned int ls, const unsigned int index) const
std::string	getOpenRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getOutputJsonFilePath (const unsigned int ls, std::string const &stream) const
std::string	getProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
std::string	getRawFilePath (const unsigned int ls, const unsigned int index) const
std::string	getRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
unsigned int	getRunNumber () const
std::string	getRunOpenDirPath () const
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 (boost::filesystem::path const &jsonSourcePath)
void	initRun ()
bool	isSingleStreamThread ()
void	lockFULocal ()
void	lockFULocal2 ()
void	lockInitLock ()
bool	outputAdler32Recheck () const
void	overrideRunNumber (unsigned int run)
void	postEndRun (edm::GlobalContext const &globalContext)
void	preallocate (edm::service::SystemBounds const &bounds)
void	preBeginJob (edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &)
void	preBeginRun (edm::GlobalContext const &globalContext)
void	preGlobalEndLumi (edm::GlobalContext const &globalContext)
int	readLastLSEntry (std::string const &file)
void	removeFile (unsigned int ls, unsigned int index)
void	removeFile (std::string)
void	setDeleteTracking (std::mutex *fileDeleteLock, std::list< std::pair< int, InputFile * >> *filesToDelete)
void	setFMS (evf::FastMonitoringService *fms)
void	tryInitializeFuLockFile ()
void	unlockFULocal ()
void	unlockFULocal2 ()
void	unlockInitLock ()
FileStatus	updateFuLock (unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint64_t &lockWaitTime)
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)

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

Private Attributes
std::string	base_dir_
std::string	bu_base_dir_
struct flock	bu_r_flk
struct flock	bu_r_fulk
FILE *	bu_r_lock_stream
int	bu_readlock_fd_
std::string	bu_run_dir_
std::string	bu_run_open_dir_
FILE *	bu_t_monitor_stream
struct flock	bu_w_flk
struct flock	bu_w_fulk
FILE *	bu_w_lock_stream
FILE *	bu_w_monitor_stream
int	bu_writelock_fd_
bool	directorBu_
DirManager	dirManager_
jsoncollector::DataPointDefinition *	dpd_
std::unique_ptr< boost::asio::ip::tcp::resolver::iterator >	endpoint_iterator_
unsigned int	eolsNFilesIndex_ = 1
std::string	fileBrokerHost_
bool	fileBrokerKeepAlive_
std::string	fileBrokerPort_
bool	fileBrokerUseLocalLock_
std::mutex *	fileDeleteLockPtr_ = nullptr
std::list< std::pair< int, InputFile * > > *	filesToDeletePtr_ = nullptr
evf::FastMonitoringService *	fms_ = nullptr
int	fu_readwritelock_fd_
struct flock	fu_rw_flk
struct flock	fu_rw_fulk
FILE *	fu_rw_lock_stream
int	fulocal_rwlock_fd2_
int	fulocal_rwlock_fd_
std::string	fulockfile_
unsigned int	fuLockPollInterval_
std::string	hltSourceDirectory_
std::string	hostname_
pthread_mutex_t	init_lock_ = PTHREAD_MUTEX_INITIALIZER
boost::asio::io_service	io_service_
tbb::concurrent_hash_map< std::string, std::string >	mergeTypeMap_
std::string	mergeTypePset_
unsigned int	nStreams_ =0
unsigned int	nThreads_ =0
bool	outputAdler32Recheck_
std::string	pid_
unsigned long	previousFileSize_
std::unique_ptr< boost::asio::ip::tcp::resolver::query >	query_
bool	readEolsDefinition_ = true
bool	requireTSPSet_
std::unique_ptr< boost::asio::ip::tcp::resolver >	resolver_
unsigned int	run_
std::string	run_dir_
std::string	run_nstring_
std::string	run_string_
std::string	selectedTransferMode_
std::unique_ptr< boost::asio::ip::tcp::socket >	socket_
unsigned int	startFromLS_ = 1
unsigned int	stop_ls_override_ = 0
std::string	stopFilePath_
std::string	stopFilePathPid_
std::shared_ptr< Json::Value >	transferSystemJson_
bool	useFileBroker_

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

Detailed Description

Definition at line 62 of file EvFDaqDirector.h.

Member Enumeration Documentation

enum evf::EvFDaqDirector::FileStatus

Enumerator
noFile
sameFile
newFile
newLumi
runEnded
runAbort

Definition at line 66 of file EvFDaqDirector.h.

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

Constructor & Destructor Documentation

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

explicit

Definition at line 39 of file EvFDaqDirector.cc.

References endpoint_iterator_, Exception, fileBrokerHost_, fileBrokerPort_, fileBrokerUseLocalLock_, fuLockPollInterval_, hostname_, recoMuon::in, io_service_, postEndRun(), preallocate(), preBeginJob(), preBeginRun(), preGlobalEndLumi(), query_, resolver_, socket_, startFromLS_, trackingPlots::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", ".")),
    directorBu_(pset.getUntrackedParameter<bool> ("directorIsBu", false)),
    run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
    useFileBroker_(pset.getUntrackedParameter<bool> ("useFileBroker", false)),
    fileBrokerHost_(pset.getUntrackedParameter<std::string>("fileBrokerHost","")),
    fileBrokerPort_(pset.getUntrackedParameter<std::string>("fileBrokerPort","8080")),
    fileBrokerKeepAlive_(pset.getUntrackedParameter<bool>("fileBrokerKeepAlive", true)),
    fileBrokerUseLocalLock_(pset.getUntrackedParameter<bool>("fileBrokerUseLocalLock",true)),
    outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck",false)),
    requireTSPSet_(pset.getUntrackedParameter<bool>("requireTransfersPSet",false)),
    selectedTransferMode_(pset.getUntrackedParameter<std::string>("selectedTransferMode","")),
    hltSourceDirectory_(pset.getUntrackedParameter<std::string>("hltSourceDirectory","")),
    fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>("fuLockPollInterval",2000)),
    mergeTypePset_(pset.getUntrackedParameter<std::string>("mergeTypePset","")),
    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 = getenv("FFF_LOCKPOLLINTERVAL");
    if (fuLockPollIntervalPtr) {
      try {
        fuLockPollInterval_=boost::lexical_cast<unsigned int>(std::string(fuLockPollIntervalPtr));
        edm::LogInfo("EvFDaqDirector") << "Setting fu lock poll interval by environment string: " << fuLockPollInterval_ << " us";
      }
      catch( boost::bad_lexical_cast const& ) {
        edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fuLockPollIntervalPtr);
      }
    }

    //override file service parameter if specified by environment
    char * fileBrokerParamPtr = getenv("FFF_USEFILEBROKER");
    if (fileBrokerParamPtr) {
      try {
        useFileBroker_=(boost::lexical_cast<unsigned int>(std::string(fileBrokerParamPtr)))>0;
        edm::LogInfo("EvFDaqDirector") << "Setting useFileBroker parameter by environment string: " << useFileBroker_;
      }
      catch( boost::bad_lexical_cast const& ) {
        edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fileBrokerParamPtr);
      }
    }
    if (useFileBroker_) {
      if (fileBrokerHost_.empty()) {
        //find BU data address from hltd configuration
        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()) {
        resolver_ = std::make_unique<boost::asio::ip::tcp::resolver>(io_service_);
        query_ = std::make_unique<boost::asio::ip::tcp::resolver::query>(fileBrokerHost_, fileBrokerPort_); //default port
        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_);
      }
      else {
        throw cms::Exception("EvFDaqDirector") << "No file service or BU data address information";
      }
    }

    char * startFromLSPtr = getenv("FFF_STARTFROMLS");
    if (startFromLSPtr) {
      try {
        startFromLS_=boost::lexical_cast<unsigned int>(std::string(startFromLSPtr));
        edm::LogInfo("EvFDaqDirector") << "Setting start from LS by environment string: " << startFromLS_;
      }
      catch( boost::bad_lexical_cast const& ) {
        edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(startFromLSPtr);
      }
    }

    //override file service parameter if specified by environment
    char * fileBrokerUseLockParamPtr = getenv("FFF_FILEBROKERUSELOCALLOCK");
    if (fileBrokerUseLockParamPtr) {
      try {
        fileBrokerUseLocalLock_=(boost::lexical_cast<unsigned int>(std::string(fileBrokerUseLockParamPtr)))>0;
        edm::LogInfo("EvFDaqDirector") << "Setting fileBrokerUseLocalLock parameter by environment string: " << fileBrokerUseLocalLock_;
      }
      catch( boost::bad_lexical_cast const& ) {
        edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fileBrokerUseLockParamPtr);
      }
    }

  }

evf::EvFDaqDirector::~EvFDaqDirector

(

)

Definition at line 311 of file EvFDaqDirector.cc.

References RecoEcal_EventContent_cff::ec, 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

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

inline

Definition at line 78 of file EvFDaqDirector.h.

Referenced by grabNextJsonFile(), and grabNextJsonFileAndUnlock().

{return run_dir_;}

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

inline

Definition at line 79 of file EvFDaqDirector.h.

{return bu_run_dir_;}

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

inline

Definition at line 80 of file EvFDaqDirector.h.

{return bu_run_open_dir_;}

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

private

Definition at line 797 of file EvFDaqDirector.cc.

References evf::FastMonitoringService::accumulateFileSize(), fms_, getEoLSFilePathOnBU(), getInputJsonFilePath(), getNFilesFromEoLS(), eostools::ls(), previousFileSize_, trackingPlots::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;
    }

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

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

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

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

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

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

    nextFile = getInputJsonFilePath(ls,0);
    if (stat(nextFile.c_str(), &buf) == 0) {
      // a new file was found at new lumisection, index 0
      previousFileSize_ = buf.st_size;
          fsize = buf.st_size;
      return true;
    }
        else {
          //change of policy: we need to cycle through each LS
          return false;
        }
        BUEoLSFile = getEoLSFilePathOnBU(ls);
        eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
      }
    }
    // no new file found
    return false;
  }

void evf::EvFDaqDirector::checkMergeTypePSet

(

edm::ProcessContext const &

pc

)

Definition at line 1750 of file EvFDaqDirector.cc.

References edm::ParameterSet::existsAs(), edm::ParameterSet::getParameterSet(), edm::getParameterSet(), mergeTypeMap_, mergeTypePset_, edm::ProcessContext::parameterSetID(), BPhysicsValidation_cfi::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 (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();
      }
    }
  }

void evf::EvFDaqDirector::checkTransferSystemPSet

(

edm::ProcessContext const &

pc

)

Definition at line 1647 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";
    }
  }

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

Definition at line 1248 of file EvFDaqDirector.cc.

References bu_run_dir_, DBConfiguration_cff::connect, MillePedeFileConverter_cfg::e, RecoEcal_EventContent_cff::ec, endpoint_iterator_, cppFunctionSkipper::exception, fileBrokerHost_, fileBrokerKeepAlive_, RecoTauValidation_cfi::header, SiStripPI::max, newFile, noFile, callgraph::path, pid_, run_nstring_, runEnded, socket_, AlCaHLTBitMon_QueryRunRegistry::string, and TriggerAnalyzer::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");


        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());
                nextFileRaw = bu_run_dir_ +"/" + filestem+".raw";//raw files are not moved
                filestem = bu_run_dir_ + "/" + fileprefix + filestem;
                nextFileJson = filestem + ".jsn";
                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;
  }

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

Definition at line 936 of file EvFDaqDirector.cc.

References getBoLSFilePathOnFU(), trackingPlots::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);
    }
  }

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

Definition at line 947 of file EvFDaqDirector.cc.

References createBoLSFile(), runEdmFileComparison::found, getEoLSFilePathOnFU(), trackingPlots::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 ) {
        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
    }
  }

void evf::EvFDaqDirector::createProcessingNotificationMaybe

(

)

const

Definition at line 1783 of file EvFDaqDirector.cc.

References run_dir_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::checkNextEvent().

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

void evf::EvFDaqDirector::createRunOpendirMaybe

(

)

Definition at line 1604 of file EvFDaqDirector.cc.

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

Referenced by initRun().

                                             {
    // create open dir if not already there

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

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

private

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

private

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

static

Definition at line 342 of file EvFDaqDirector.cc.

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

  {
    edm::ParameterSetDescription desc;
    desc.setComment("Service used for file locking arbitration and for propagating information between other EvF components");
    desc.addUntracked<std::string> ("baseDir", ".")->setComment("Local base directory for run output");
    desc.addUntracked<std::string> ("buBaseDir", ".")->setComment("BU base ramdisk directory ");
    desc.addUntracked<unsigned int> ("runNumber",0)->setComment("Run Number in ramdisk to open");
    desc.addUntracked<bool> ("useFileBroker", false)->setComment("Use BU file service to grab input data instead of NFS file locking");
    desc.addUntracked<std::string> ("fileBrokerHost", "")->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<bool>("outputAdler32Recheck",false)->setComment("Check Adler32 of per-process output files while micro-merging");
    desc.addUntracked<bool>("requireTransfersPSet",false)->setComment("Require complete transferSystem PSet in the process configuration");
    desc.addUntracked<std::string>("selectedTransferMode","")->setComment("Selected transfer mode (choice in Lvl0 propagated as Python parameter");
    desc.addUntracked<unsigned int>("fuLockPollInterval",2000)->setComment("Lock polling interval in microseconds for the input directory file lock");
    desc.addUntracked<std::string>("mergingPset","")->setComment("Name of merging PSet to look for merging type definitions for streams");
    desc.setAllowAnything();
    descriptions.add("EvFDaqDirector", desc);
  }

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

inline

Definition at line 83 of file EvFDaqDirector.h.

References eostools::ls(), and AlCaHLTBitMon_QueryRunRegistry::string.

{ return dirManager_.findRunDir(run_);}

std::string evf::EvFDaqDirector::getBoLSFilePathOnFU

(

const unsigned int

ls

)

const

Definition at line 514 of file EvFDaqDirector.cc.

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

Referenced by createBoLSFile().

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

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

Definition at line 450 of file EvFDaqDirector.cc.

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

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

std::string evf::EvFDaqDirector::getEoLSFilePathOnBU

(

const unsigned int

ls

)

const

Definition at line 506 of file EvFDaqDirector.cc.

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

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

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

std::string evf::EvFDaqDirector::getEoLSFilePathOnFU

(

const unsigned int

ls

)

const

Definition at line 510 of file EvFDaqDirector.cc.

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

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

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

std::string evf::EvFDaqDirector::getEoRFilePath

(

)

const

Definition at line 518 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

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

std::string evf::EvFDaqDirector::getEoRFilePathOnFU

(

)

const

Definition at line 522 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::checkNextEvent().

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

std::string evf::EvFDaqDirector::getFFFParamsFilePathOnBU

(

)

const

Definition at line 526 of file EvFDaqDirector.cc.

References bu_run_dir_.

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

std::string evf::EvFDaqDirector::getInitFilePath

(

std::string const &

stream

)

const

Definition at line 478 of file EvFDaqDirector.cc.

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

Referenced by DQMFileSaver::globalBeginRun().

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

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

Definition at line 434 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

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

unsigned int evf::EvFDaqDirector::getLumisectionToStart

(

)

const

Definition at line 1631 of file EvFDaqDirector.cc.

References run_dir_, run_string_, startFromLS_, trackingPlots::stat, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::readSupervisor().

                                                           {
      std::string fileprefix = run_dir_ +"/"+run_string_+"_ls";
      std::string fullpath;
      struct stat buf;
      unsigned int lscount=startFromLS_;
      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;
  }

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

Definition at line 470 of file EvFDaqDirector.cc.

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

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

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

Definition at line 466 of file EvFDaqDirector.cc.

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

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

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

Definition at line 490 of file EvFDaqDirector.cc.

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

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

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

Definition at line 502 of file EvFDaqDirector.cc.

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

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

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

Definition at line 1490 of file EvFDaqDirector.cc.

References bu_run_dir_, contactFileBroker(), createLumiSectionFiles(), fileBrokerUseLocalLock_, grabNextJsonFile(), mps_fire::i, lockFULocal2(), eostools::ls(), SiStripPI::max, newFile, noFile, readLastLSEntry(), runEnded, trackingPlots::stat, stop_ls_override_, stopFilePath_, stopFilePathPid_, AlCaHLTBitMon_QueryRunRegistry::string, mitigatedMETSequence_cff::U, and unlockFULocal2().

Referenced by FedRawDataInputSource::readSupervisor().

                                                                                                                                         {

    EvFDaqDirector::FileStatus fileStatus = noFile;

    //int retval = -1;
    //int lock_attempts = 0;
    //long total_lock_attempts = 0;

    struct stat buf;
    int stopFileLS = -1;
    int stopFileCheck = stat(stopFilePath_.c_str(),&buf);
    int stopFilePidCheck = stat(stopFilePathPid_.c_str(),&buf);
    if (stopFileCheck==0 || stopFilePidCheck==0) {
        if (stopFileCheck==0)
          stopFileLS = readLastLSEntry(stopFilePath_);
        else
          stopFileLS = 1;//stop without drain if only pid is stopped
        if (!stop_ls_override_) {
          //if lumisection is higher than in stop file, should quit at next from current
          if (stopFileLS>=0 && (int)ls>=stopFileLS) stopFileLS = stop_ls_override_ = ls;
        }
        else stopFileLS = stop_ls_override_;
        edm::LogWarning("EvFDaqDirector") << "Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
        //return runEnded;
    }
    else //if file was removed before reaching stop condition, reset this
      stop_ls_override_ = 0;

    /* look for EoLS
    if (stat(getEoLSFilePathOnFU(currentLumiSection).c_str(),&buf)==0) {
      edm::LogWarning("EvFDaqDirector") << "Detected local EoLS for lumisection "<< currentLumiSection ; 
      ls++;
      return noFile;
    }
    */


    timeval ts_lockbegin;
    gettimeofday(&ts_lockbegin,nullptr);

    std::string nextFileJson;
    uint32_t serverLS, closedServerLS;
    unsigned int serverHttpStatus;
    bool serverError;

    //local lock to force index json and EoLS files to appear in order
    if (fileBrokerUseLocalLock_)
      lockFULocal2();

    int maxLS = stopFileLS < 0 ? -1: std::max(stopFileLS,(int)currentLumiSection);
    fileStatus = contactFileBroker(serverHttpStatus, serverError, serverLS, closedServerLS, nextFileJson, nextFileRaw, maxLS);

    if (serverError) {
      //do not update anything
      if (fileBrokerUseLocalLock_)
        unlockFULocal2();
      return noFile;
    }

    //handle creation of EoLS and BoLS files if lumisection has changed
    if (currentLumiSection==0) {
        if (fileStatus == runEnded) {
          createLumiSectionFiles(closedServerLS,0);
          createLumiSectionFiles(serverLS,closedServerLS,false); // +1
        }
        else 
          createLumiSectionFiles(serverLS,0);
    } else {
        //loop over and create any EoLS files missing
        if (closedServerLS>=currentLumiSection) {
          for (uint32_t i=std::max(currentLumiSection,1U);i<=closedServerLS;i++) 
            createLumiSectionFiles(i+1,i);
          }
    }

    bool fileFound=true;

    if (fileStatus == newFile)
      serverEventsInNewFile = grabNextJsonFile(nextFileJson,nextFileRaw,fileSizeFromJson,fileFound);

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

    //can unlock because all files have been created locally
    if (fileBrokerUseLocalLock_)
      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;
  }

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

private

Definition at line 736 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

                                                            {

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

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

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

    //read definition 
    if (readEolsDefinition_) {
      //std::string def = boost::algorithm::trim(dp.getDefinition());
      std::string def = dp.getDefinition();
      if (def.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 boost::lexical_cast<int>(data);
  }

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

Definition at line 454 of file EvFDaqDirector.cc.

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

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

std::string evf::EvFDaqDirector::getOpenInitFilePath

(

std::string const &

stream

)

const

Definition at line 474 of file EvFDaqDirector.cc.

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

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

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

Definition at line 446 of file EvFDaqDirector.cc.

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

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

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

Definition at line 458 of file EvFDaqDirector.cc.

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

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

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

Definition at line 482 of file EvFDaqDirector.cc.

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

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

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

Definition at line 442 of file EvFDaqDirector.cc.

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

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

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

Definition at line 494 of file EvFDaqDirector.cc.

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

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

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

Definition at line 462 of file EvFDaqDirector.cc.

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

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

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

Definition at line 486 of file EvFDaqDirector.cc.

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

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

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

Definition at line 438 of file EvFDaqDirector.cc.

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

Referenced by removeFile().

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

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

Definition at line 498 of file EvFDaqDirector.cc.

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

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

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

inline

Definition at line 115 of file EvFDaqDirector.h.

{ return run_; }

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

inline

Definition at line 108 of file EvFDaqDirector.h.

Referenced by createRunOpendirMaybe(), and initRun().

{return run_dir_ +"/open";}

std::string evf::EvFDaqDirector::getStreamDestinations

(

std::string const &

stream

)

const

Definition at line 1703 of file EvFDaqDirector.cc.

References Json::Value::begin(), Json::Value::end(), Exception, mps_check::msg, requireTSPSet_, selectedTransferMode_, AlCaHLTBitMon_QueryRunRegistry::string, and transferSystemJson_.

  {
    std::string streamRequestName;
    if (transferSystemJson_->isMember(stream.c_str()))
      streamRequestName = stream;
    else {
      std::stringstream msg;
      msg << "Transfer system mode definitions missing for -: " << stream;
      if (requireTSPSet_)
        throw cms::Exception("EvFDaqDirector") << msg.str();
      else {
        edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
        return std::string("Failsafe");
      }
    }
    //return empty if strict check parameter is not on
    if (!requireTSPSet_ && (selectedTransferMode_.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;
  }

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

Definition at line 1768 of file EvFDaqDirector.cc.

References mergeTypeMap_, MergeTypeNames_, 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;
  }

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

Definition at line 964 of file EvFDaqDirector.cc.

References baseRunDir(), data, jsoncollector::DataPoint::deserialize(), dpd_, MillePedeFileConverter_cfg::e, cppFunctionSkipper::exception, jsoncollector::DataPoint::getData(), Json::Reader::getFormatedErrorMessages(), jsoncollector::DataPointDefinition::getNames(), mps_fire::i, LogDebug, lumiCalc2::outfile, Json::Reader::parse(), callgraph::path, pid_, matplotRender::reader, mps_fire::result, trackingPlots::stat, AlCaHLTBitMon_QueryRunRegistry::string, summarizeEdmComparisonLogfiles::success, and TriggerAnalyzer::write().

Referenced by getNextFromFileBroker().

  {
    fileFound=true;

    //should be ported to use fffnaming
    std::ostringstream fileNameWithPID;
    fileNameWithPID << boost::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);
      usleep(100000);
      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);
      usleep(100000);
      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 {
          boost::filesystem::ifstream ij(jsonDestPath);
          ss << ij.rdbuf();
        }
        catch (boost::filesystem::filesystem_error const& ex) {
          edm::LogError("EvFDaqDirector") << "grabNextJsonFile - BOOST 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()) {
        data = dp.getData()[i];
            try {
              fileSizeFromJson = boost::lexical_cast<long>(data);
        }
            catch( boost::bad_lexical_cast const& ) {
              //non-fatal currently, processing can continue without this value
              edm::LogWarning("EvFDaqDirector") << "grabNextJsonFile - error parsing number of Bytes from BU JSON. "
                                                << "Input value is -: " << data;
            }
            break;
          }
        }
      }
      return boost::lexical_cast<int>(data);
    }
    catch( boost::bad_lexical_cast const& e) {
      edm::LogError("EvFDaqDirector") << "grabNextJsonFile - error parsing number of 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;
  }

int evf::EvFDaqDirector::grabNextJsonFileAndUnlock

(

boost::filesystem::path const &

jsonSourcePath

)

Definition at line 1144 of file EvFDaqDirector.cc.

References baseRunDir(), popcon2dropbox::copy(), data, jsoncollector::DataPoint::deserialize(), dpd_, MillePedeFileConverter_cfg::e, cppFunctionSkipper::exception, Exception, jsoncollector::DataPoint::getData(), Json::Reader::getFormatedErrorMessages(), jsoncollector::DataPointDefinition::getNames(), mps_fire::i, LogDebug, Json::Reader::parse(), callgraph::path, matplotRender::reader, MatrixUtil::remove(), AlCaHLTBitMon_QueryRunRegistry::string, summarizeEdmComparisonLogfiles::success, and unlockFULocal().

Referenced by FedRawDataInputSource::readSupervisor().

  {
    std::string data;
    try {
      // assemble json destination path
      boost::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 {
        boost::filesystem::copy(jsonSourcePath,jsonDestPath);
      }
      catch (boost::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);
        boost::filesystem::copy(jsonSourcePath,jsonDestPath);
      }
      unlockFULocal();

      try {
        //sometimes this fails but file gets deleted
        boost::filesystem::remove(jsonSourcePath);
      }
      catch (boost::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();
      }

      boost::filesystem::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 boost::lexical_cast<int>(data);
    }
    catch (boost::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( boost::bad_lexical_cast const& ) {
      edm::LogError("EvFDaqDirector") << "grabNextFile error parsing number of 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;
  }

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

private

void evf::EvFDaqDirector::initRun

(

)

Definition at line 152 of file EvFDaqDirector.cc.

References base_dir_, bu_base_dir_, bu_run_dir_, bu_run_open_dir_, bu_w_lock_stream, bu_writelock_fd_, eostools::chmod(), createRunOpendirMaybe(), directorBu_, dpd_, Exception, fu_readwritelock_fd_, fu_rw_lock_stream, fulocal_rwlock_fd2_, fulocal_rwlock_fd_, fulockfile_, getRunOpenDirPath(), hltSourceDirectory_, init_lock_, eostools::mkdir(), openFULockfileStream(), pid_, run_, run_dir_, run_nstring_, run_string_, trackingPlots::stat, stopFilePath_, stopFilePathPid_, AlCaHLTBitMon_QueryRunRegistry::string, and tryInitializeFuLockFile().

Referenced by preallocate().

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

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

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

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

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

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

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

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

    // BU INITIALIZES LOCK FILE
    // FU LOCK FILE OPEN
    openFULockfileStream(true);
    tryInitializeFuLockFile();
    fflush(fu_rw_lock_stream);
    close(fu_readwritelock_fd_);

        if (!hltSourceDirectory_.empty())
    {
      struct stat buf;
      if (stat(hltSourceDirectory_.c_str(),&buf)==0) {
        std::string hltdir=bu_run_dir_+"/hlt";
        std::string tmphltdir=bu_run_open_dir_+"/hlt";
        retval = mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
        if (retval != 0 && errno != EEXIST)
          throw cms::Exception("DaqDirector")
            << " Error creating bu run dir -: " << hltdir
            << " mkdir error:" << strerror(errno) << "\n";

            boost::filesystem::copy_file(hltSourceDirectory_+"/HltConfig.py",tmphltdir+"/HltConfig.py");

            boost::filesystem::copy_file(hltSourceDirectory_+"/fffParameters.jsn",tmphltdir+"/fffParameters.jsn");

            boost::filesystem::rename(tmphltdir,hltdir);
      }
          else
        throw cms::Exception("DaqDirector") << " Error looking for HLT configuration -: " << hltSourceDirectory_;
    }
        //else{}//no configuration specified
      }
    else
    {
      // for FU, check if bu base dir exists

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

    bu_run_dir_ = bu_base_dir_ + "/" + run_string_;
    fulockfile_ = bu_run_dir_ + "/fu.lock";
    openFULockfileStream(false);
      }

    pthread_mutex_init(&init_lock_,nullptr);

    stopFilePath_ = run_dir_+"/CMSSW_STOP";
    std::stringstream sstp;
    sstp << stopFilePath_ << "_pid" << pid_;
    stopFilePathPid_ = sstp.str();


    if (!directorBu_) {
      std::string defPath = bu_run_dir_ + "/jsd/rawData.jsd";
      struct stat statbuf;
      if (!stat(defPath.c_str(), &statbuf)) 
        edm::LogInfo("EvFDaqDirector") << "found JSD file in ramdisk -: " << defPath;
      else {
        //look in source directory if not present in ramdisk
        std::string defPathSuffix = "src/EventFilter/Utilities/plugins/budef.jsd";
        defPath = std::string(getenv("CMSSW_BASE")) + "/" + defPathSuffix;
        if (stat(defPath.c_str(), &statbuf)) {
          defPath = std::string(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);
    }

  }

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

private

bool evf::EvFDaqDirector::isSingleStreamThread ( )

inline

Definition at line 119 of file EvFDaqDirector.h.

References FrontierConditions_GlobalTag_cff::file, callgraph::path, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::getNextEvent().

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

void evf::EvFDaqDirector::lockFULocal

(

)

Definition at line 915 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by updateFuLock().

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

void evf::EvFDaqDirector::lockFULocal2

(

)

Definition at line 926 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

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

void evf::EvFDaqDirector::lockInitLock

(

)

Definition at line 907 of file EvFDaqDirector.cc.

References init_lock_.

                                    {
    pthread_mutex_lock(&init_lock_);
  }

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

static

Definition at line 1789 of file EvFDaqDirector.cc.

References sysUtil::pid, and command_line::start.

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

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

private

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

private

Definition at line 886 of file EvFDaqDirector.cc.

References eostools::chmod(), 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);

  }

bool evf::EvFDaqDirector::outputAdler32Recheck ( ) const

inline

Definition at line 109 of file EvFDaqDirector.h.

References AlCaHLTBitMon_QueryRunRegistry::string.

{return outputAdler32Recheck_;}

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

private

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

inline

Definition at line 77 of file EvFDaqDirector.h.

References writedatasetfile::run.

{run_=run;}

void evf::EvFDaqDirector::postEndRun

(

edm::GlobalContext const &

globalContext

)

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

void evf::EvFDaqDirector::preallocate

(

edm::service::SystemBounds const &

bounds

)

Definition at line 334 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

                                                                       {

    initRun();

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

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

Definition at line 363 of file EvFDaqDirector.cc.

References checkMergeTypePSet(), and checkTransferSystemPSet().

Referenced by EvFDaqDirector().

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

void evf::EvFDaqDirector::preBeginRun

(

edm::GlobalContext const &

globalContext

)

Definition at line 369 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

                                                                        {

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

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

void evf::EvFDaqDirector::preGlobalEndLumi

(

edm::GlobalContext const &

globalContext

)

Definition at line 390 of file EvFDaqDirector.cc.

References cppFunctionSkipper::exception, fileDeleteLockPtr_, filesToDeletePtr_, LogDebug, eostools::ls(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), callgraph::path, and MatrixUtil::remove().

Referenced by EvFDaqDirector().

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

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

int evf::EvFDaqDirector::readLastLSEntry

(

std::string const &

file

)

Definition at line 1615 of file EvFDaqDirector.cc.

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

Referenced by getNextFromFileBroker(), and updateFuLock().

                                                           {

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

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

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

  }

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

Definition at line 537 of file EvFDaqDirector.cc.

References getRawFilePath().

Referenced by postEndRun().

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

void evf::EvFDaqDirector::removeFile

(

std::string

filename

)

Definition at line 530 of file EvFDaqDirector.cc.

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

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

inline

Definition at line 139 of file EvFDaqDirector.h.

References beamerCreator::create(), fffnaming::eolsFileName(), fffnaming::eorFileName(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

                                                                                                        {
        fileDeleteLockPtr_=fileDeleteLock;
        filesToDeletePtr_ = filesToDelete;
      }

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

inline

Definition at line 118 of file EvFDaqDirector.h.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

{fms_=fms;}

void evf::EvFDaqDirector::tryInitializeFuLockFile

(

)

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

void evf::EvFDaqDirector::unlockFULocal

(

)

Definition at line 920 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

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

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

void evf::EvFDaqDirector::unlockFULocal2

(

)

Definition at line 931 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

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

void evf::EvFDaqDirector::unlockInitLock

(

)

Definition at line 911 of file EvFDaqDirector.cc.

References init_lock_.

                                      {
    pthread_mutex_unlock(&init_lock_);
  }

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

Definition at line 541 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::readSupervisor().

                                                                                                                                      {
    EvFDaqDirector::FileStatus fileStatus = noFile;

    int retval = -1;
    int lock_attempts = 0;
    long total_lock_attempts = 0;

    struct stat buf;
    int stopFileLS = -1;
    int stopFileCheck = stat(stopFilePath_.c_str(),&buf);
    int stopFilePidCheck = stat(stopFilePathPid_.c_str(),&buf);
    if (stopFileCheck==0 || stopFilePidCheck==0) {
        if (stopFileCheck==0)
          stopFileLS = readLastLSEntry(stopFilePath_);
        else
          stopFileLS = 1;//stop without drain if only pid is stopped
        if (!stop_ls_override_) {
          //if lumisection is higher than in stop file, should quit at next from current
          if (stopFileLS>=0 && (int)ls>=stopFileLS) stopFileLS = stop_ls_override_ = ls;
        }
        else stopFileLS = stop_ls_override_;
        edm::LogWarning("EvFDaqDirector") << "Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
        //return runEnded;
    }
    else //if file was removed before reaching stop condition, reset this
      stop_ls_override_ = 0;

    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, stopFileLS);
          //avoid 2 lumisections jump
          if (ls && readLs>currentLs && currentLs > ls) {
            ls++;
            readLs=currentLs=ls;
            readIndex=0;
            bumpedOk=false;
            //no write to lock file
          }
          else {
            if (ls==0 && readLs>currentLs) {
              //make sure to intialize always with LS found in the lock file, with possibility of grabbing index file immediately
              //in this case there is no new file in the same LS
              //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 {
            throw cms::Exception("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error " << strerror(errno);
          }
    }
        else if (currentLs < readLs) {
          //there is no new file in next LS (yet), but lock file can be updated to the next LS
      check = fseek(fu_rw_lock_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 {
            throw cms::Exception("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error " << strerror(errno);
          }
        }
      } else {
    edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for reading failed with error " << strerror(errno);
      }
    } else {
      edm::LogError("EvFDaqDirector") << "fu read/write lock stream is invalid " << strerror(errno);
    }
    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;
  }

bool evf::EvFDaqDirector::useFileBroker ( ) const

inline

Definition at line 81 of file EvFDaqDirector.h.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

{return useFileBroker_;}

Member Data Documentation

std::string evf::EvFDaqDirector::base_dir_

private

Definition at line 160 of file EvFDaqDirector.h.

Referenced by initRun().

std::string evf::EvFDaqDirector::bu_base_dir_

private

Definition at line 161 of file EvFDaqDirector.h.

Referenced by initRun().

struct flock evf::EvFDaqDirector::bu_r_flk

private

Definition at line 203 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_r_fulk

private

Definition at line 205 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_r_lock_stream

private

Definition at line 193 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_readlock_fd_

private

Definition at line 186 of file EvFDaqDirector.h.

Referenced by postEndRun().

std::string evf::EvFDaqDirector::bu_run_dir_

private

Definition at line 182 of file EvFDaqDirector.h.

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

std::string evf::EvFDaqDirector::bu_run_open_dir_

private

Definition at line 183 of file EvFDaqDirector.h.

Referenced by initRun().

FILE* evf::EvFDaqDirector::bu_t_monitor_stream

private

Definition at line 196 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_flk

private

Definition at line 202 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_fulk

private

Definition at line 204 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_w_lock_stream

private

Definition at line 192 of file EvFDaqDirector.h.

Referenced by initRun().

FILE* evf::EvFDaqDirector::bu_w_monitor_stream

private

Definition at line 195 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_writelock_fd_

private

Definition at line 187 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

bool evf::EvFDaqDirector::directorBu_

private

Definition at line 162 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

DirManager evf::EvFDaqDirector::dirManager_

private

Definition at line 198 of file EvFDaqDirector.h.

Referenced by preBeginRun().

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

private

Definition at line 233 of file EvFDaqDirector.h.

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

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

private

Definition at line 238 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), and EvFDaqDirector().

unsigned int evf::EvFDaqDirector::eolsNFilesIndex_ = 1

private

Definition at line 220 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

std::string evf::EvFDaqDirector::fileBrokerHost_

private

Definition at line 165 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), and EvFDaqDirector().

bool evf::EvFDaqDirector::fileBrokerKeepAlive_

private

Definition at line 167 of file EvFDaqDirector.h.

Referenced by contactFileBroker().

std::string evf::EvFDaqDirector::fileBrokerPort_

private

Definition at line 166 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

bool evf::EvFDaqDirector::fileBrokerUseLocalLock_

private

Definition at line 168 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and getNextFromFileBroker().

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

private

Definition at line 211 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi().

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

private

Definition at line 212 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi().

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

private

Definition at line 209 of file EvFDaqDirector.h.

Referenced by bumpFile().

int evf::EvFDaqDirector::fu_readwritelock_fd_

private

Definition at line 188 of file EvFDaqDirector.h.

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

struct flock evf::EvFDaqDirector::fu_rw_flk

private

Definition at line 206 of file EvFDaqDirector.h.

Referenced by updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_fulk

private

Definition at line 207 of file EvFDaqDirector.h.

Referenced by updateFuLock().

FILE* evf::EvFDaqDirector::fu_rw_lock_stream

private

Definition at line 194 of file EvFDaqDirector.h.

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

int evf::EvFDaqDirector::fulocal_rwlock_fd2_

private

Definition at line 190 of file EvFDaqDirector.h.

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

int evf::EvFDaqDirector::fulocal_rwlock_fd_

private

Definition at line 189 of file EvFDaqDirector.h.

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

std::string evf::EvFDaqDirector::fulockfile_

private

Definition at line 184 of file EvFDaqDirector.h.

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

unsigned int evf::EvFDaqDirector::fuLockPollInterval_

private

Definition at line 174 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateFuLock().

std::string evf::EvFDaqDirector::hltSourceDirectory_

private

Definition at line 173 of file EvFDaqDirector.h.

Referenced by initRun().

std::string evf::EvFDaqDirector::hostname_

private

Definition at line 177 of file EvFDaqDirector.h.

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

pthread_mutex_t evf::EvFDaqDirector::init_lock_ = PTHREAD_MUTEX_INITIALIZER

private

Definition at line 214 of file EvFDaqDirector.h.

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

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

private

Definition at line 235 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

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

private

Definition at line 226 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet(), and getStreamMergeType().

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

staticprivate

Definition at line 229 of file EvFDaqDirector.h.

Referenced by getStreamMergeType().

std::string evf::EvFDaqDirector::mergeTypePset_

private

Definition at line 175 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet().

unsigned int evf::EvFDaqDirector::nStreams_ =0

private

Definition at line 216 of file EvFDaqDirector.h.

Referenced by preallocate().

unsigned int evf::EvFDaqDirector::nThreads_ =0

private

Definition at line 217 of file EvFDaqDirector.h.

Referenced by preallocate().

bool evf::EvFDaqDirector::outputAdler32Recheck_

private

Definition at line 170 of file EvFDaqDirector.h.

std::string evf::EvFDaqDirector::pid_

private

Definition at line 180 of file EvFDaqDirector.h.

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

unsigned long evf::EvFDaqDirector::previousFileSize_

private

Definition at line 200 of file EvFDaqDirector.h.

Referenced by bumpFile().

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

private

Definition at line 237 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

bool evf::EvFDaqDirector::readEolsDefinition_ = true

private

Definition at line 219 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

bool evf::EvFDaqDirector::requireTSPSet_

private

Definition at line 171 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().

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

private

Definition at line 236 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

unsigned int evf::EvFDaqDirector::run_

private

Definition at line 163 of file EvFDaqDirector.h.

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

std::string evf::EvFDaqDirector::run_dir_

private

Definition at line 181 of file EvFDaqDirector.h.

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

std::string evf::EvFDaqDirector::run_nstring_

private

Definition at line 179 of file EvFDaqDirector.h.

Referenced by contactFileBroker(), and initRun().

std::string evf::EvFDaqDirector::run_string_

private

Definition at line 178 of file EvFDaqDirector.h.

Referenced by getLumisectionToStart(), and initRun().

std::string evf::EvFDaqDirector::selectedTransferMode_

private

Definition at line 172 of file EvFDaqDirector.h.

Referenced by getStreamDestinations().

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

private

Definition at line 239 of file EvFDaqDirector.h.

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

unsigned int evf::EvFDaqDirector::startFromLS_ = 1

private

Definition at line 169 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and getLumisectionToStart().

unsigned int evf::EvFDaqDirector::stop_ls_override_ = 0

private

Definition at line 223 of file EvFDaqDirector.h.

Referenced by getNextFromFileBroker(), and updateFuLock().

std::string evf::EvFDaqDirector::stopFilePath_

private

Definition at line 221 of file EvFDaqDirector.h.

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

std::string evf::EvFDaqDirector::stopFilePathPid_

private

Definition at line 222 of file EvFDaqDirector.h.

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

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

private

Definition at line 225 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().

bool evf::EvFDaqDirector::useFileBroker_

private

Definition at line 164 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().