CMS 3D CMS Logo

 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Pages
List of all members | Public Types | Public Member Functions | Static Public Member Functions | Private Member Functions | Private Attributes | Static Private Attributes
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)
 
void createProcessingNotificationMaybe () const
 
void createRunOpendirMaybe ()
 
bool emptyLumisectionMode () const
 
 EvFDaqDirector (const edm::ParameterSet &pset, edm::ActivityRegistry &reg)
 
std::string findCurrentRunDir ()
 
std::string getBoLSFilePathOnFU (const unsigned int ls) const
 
std::string getDatFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getEoLSFilePathOnBU (const unsigned int ls) const
 
std::string getEoLSFilePathOnFU (const unsigned int ls) const
 
std::string getEoRFilePath () const
 
std::string getEoRFilePathOnFU () const
 
std::string getInitFilePath (std::string const &stream) const
 
std::string getInputJsonFilePath (const unsigned int ls, const unsigned int index) const
 
std::string getMergedDatChecksumFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getMergedDatFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getMergedProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getMergedRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getOpenDatFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getOpenInitFilePath (std::string const &stream) const
 
std::string getOpenInputJsonFilePath (const unsigned int ls, const unsigned int index) const
 
std::string getOpenOutputJsonFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getOpenProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getOpenRawFilePath (const unsigned int ls, const unsigned int index) const
 
std::string getOpenRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getOutputJsonFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getProtocolBufferHistogramFilePath (const unsigned int ls, std::string const &stream) const
 
std::string getRawFilePath (const unsigned int ls, const unsigned int index) const
 
std::string getRootHistogramFilePath (const unsigned int ls, std::string const &stream) const
 
unsigned int getRunNumber () const
 
std::string getRunOpenDirPath () const
 
std::string getStreamDestinations (std::string const &stream) const
 
std::vector< int > * getStreamFileTracker ()
 
std::string getStreamMergeType (std::string const &stream, MergeType defaultType)
 
void initRun ()
 
bool isSingleStreamThread ()
 
void lockFULocal ()
 
void lockFULocal2 ()
 
void lockInitLock ()
 
FILE * maybeCreateAndLockFileHeadForStream (unsigned int ls, std::string &stream)
 
bool microMergeDisabled () const
 
bool outputAdler32Recheck () const
 
void overrideRunNumber (unsigned int run)
 
void postEndRun (edm::GlobalContext const &globalContext)
 
void preallocate (edm::service::SystemBounds const &bounds)
 
void preBeginJob (edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &)
 
void preBeginRun (edm::GlobalContext const &globalContext)
 
void preGlobalEndLumi (edm::GlobalContext const &globalContext)
 
void preSourceEvent (edm::StreamID const &streamID)
 
int readLastLSEntry (std::string const &file)
 
void removeFile (unsigned int ls, unsigned int index)
 
void removeFile (std::string)
 
void setDeleteTracking (std::mutex *fileDeleteLock, std::list< std::pair< int, InputFile * >> *filesToDelete)
 
void setFMS (evf::FastMonitoringService *fms)
 
void tryInitializeFuLockFile ()
 
void unlockAndCloseMergeStream ()
 
void unlockFULocal ()
 
void unlockFULocal2 ()
 
void unlockInitLock ()
 
void updateFileIndex (int const &fileIndex)
 
FileStatus updateFuLock (unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint64_t &lockWaitTime)
 
 ~EvFDaqDirector ()
 

Static Public Member Functions

static void fillDescriptions (edm::ConfigurationDescriptions &descriptions)
 

Private Member Functions

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

Private Attributes

std::string base_dir_
 
std::string bu_base_dir_
 
struct flock bu_r_flk
 
struct flock bu_r_fulk
 
FILE * bu_r_lock_stream
 
int bu_readlock_fd_
 
std::string bu_run_dir_
 
std::string bu_run_open_dir_
 
FILE * bu_t_monitor_stream
 
struct flock bu_w_flk
 
struct flock bu_w_fulk
 
FILE * bu_w_lock_stream
 
FILE * bu_w_monitor_stream
 
int bu_writelock_fd_
 
int currentFileIndex_ = -1
 
int data_readwrite_fd_
 
struct flock data_rw_flk
 
struct flock data_rw_fulk
 
FILE * data_rw_stream
 
bool directorBu_
 
DirManager dirManager_
 
bool emptyLumisectionMode_
 
unsigned int eolsNFilesIndex_ = 1
 
std::mutexfileDeleteLockPtr_ = nullptr
 
std::list< std::pair< int,
InputFile * > > * 
filesToDeletePtr_ = nullptr
 
evf::FastMonitoringServicefms_ = nullptr
 
int fu_readwritelock_fd_
 
struct flock fu_rw_flk
 
struct flock fu_rw_fulk
 
FILE * fu_rw_lock_stream
 
int fulocal_rwlock_fd2_
 
int fulocal_rwlock_fd_
 
unsigned int fuLockPollInterval_
 
std::string hltSourceDirectory_
 
std::string hostname_
 
pthread_mutex_t init_lock_ = PTHREAD_MUTEX_INITIALIZER
 
std::map< std::string,
std::string > 
mergeTypeMap_
 
std::string mergeTypePset_
 
bool microMergeDisabled_
 
unsigned int nStreams_ =0
 
unsigned int nThreads_ =0
 
bool outputAdler32Recheck_
 
unsigned long previousFileSize_
 
bool readEolsDefinition_ = true
 
bool requireTSPSet_
 
unsigned int run_
 
std::string run_dir_
 
std::string run_string_
 
std::string selectedTransferMode_
 
unsigned int stop_ls_override_ = 0
 
std::string stopFilePath_
 
std::vector< int > streamFileTracker_
 
std::shared_ptr< Json::ValuetransferSystemJson_
 

Static Private Attributes

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

Detailed Description

Definition at line 54 of file EvFDaqDirector.h.

Member Enumeration Documentation

Constructor & Destructor Documentation

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

Definition at line 47 of file EvFDaqDirector.cc.

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

48  :
49  base_dir_(pset.getUntrackedParameter<std::string> ("baseDir", ".")),
50  bu_base_dir_(pset.getUntrackedParameter<std::string> ("buBaseDir", ".")),
51  directorBu_(pset.getUntrackedParameter<bool> ("directorIsBu", false)),
52  run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
53  outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck",false)),
54  requireTSPSet_(pset.getUntrackedParameter<bool>("requireTransfersPSet",false)),
55  selectedTransferMode_(pset.getUntrackedParameter<std::string>("selectedTransferMode","")),
56  hltSourceDirectory_(pset.getUntrackedParameter<std::string>("hltSourceDirectory","")),
57  fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>("fuLockPollInterval",2000)),
58  emptyLumisectionMode_(pset.getUntrackedParameter<bool>("emptyLumisectionMode",true)),
59  microMergeDisabled_(pset.getUntrackedParameter<bool>("microMergeDisabled",true)),
60  mergeTypePset_(pset.getUntrackedParameter<std::string>("mergeTypePset","")),
61  hostname_(""),
62  bu_readlock_fd_(-1),
63  bu_writelock_fd_(-1),
68 
72  //bu_w_monitor_stream(0),
73  //bu_t_monitor_stream(0),
74  data_rw_stream(0),
75 
77 
79 
80  bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
81  bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
82  bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
83  bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
84  fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
85  fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
86  data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
87  data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
88  //fulocal_rw_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
89  //fulocal_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
90  //fulocal_rw_flk2( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
91  //fulocal_rw_fulk2( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
92  {
93 
100 
101  //save hostname for later
102  char hostname[33];
103  gethostname(hostname,32);
104  hostname_ = hostname;
105 
106  char * fuLockPollIntervalPtr = getenv("FFF_LOCKPOLLINTERVAL");
107  if (fuLockPollIntervalPtr) {
108  try {
109  fuLockPollInterval_=boost::lexical_cast<unsigned int>(std::string(fuLockPollIntervalPtr));
110  edm::LogInfo("EvFDaqDirector") << "Setting fu lock poll interval by environment string: " << fuLockPollInterval_ << " us";
111  }
112  catch( boost::bad_lexical_cast const& ) {
113  edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fuLockPollIntervalPtr);
114  }
115  }
116 
117  char * emptyLumiModePtr = getenv("FFF_EMPTYLSMODE");
118  if (emptyLumiModePtr) {
119  emptyLumisectionMode_ = true;
120  edm::LogInfo("EvFDaqDirector") << "Setting empty lumisection mode";
121  }
122 
123  char * microMergeDisabledPtr = getenv("FFF_MICROMERGEDISABLED");
124  if (microMergeDisabledPtr) {
125  microMergeDisabled_ = true;
126  edm::LogInfo("EvFDaqDirector") << "Disabling dat file micro-merge by the HLT process (delegated to the hlt daemon)";
127  }
128  }
struct flock bu_w_fulk
T getUntrackedParameter(std::string const &, T const &) const
struct flock fu_rw_flk
void watchPreallocate(Preallocate::slot_type const &iSlot)
void watchPreGlobalEndLumi(PreGlobalEndLumi::slot_type const &iSlot)
struct flock bu_r_fulk
struct flock data_rw_flk
unsigned long previousFileSize_
struct flock fu_rw_fulk
std::string hltSourceDirectory_
std::string mergeTypePset_
std::string bu_base_dir_
std::string selectedTransferMode_
void preBeginJob(edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &)
void watchPreGlobalBeginRun(PreGlobalBeginRun::slot_type const &iSlot)
struct flock bu_w_flk
void preBeginRun(edm::GlobalContext const &globalContext)
void preSourceEvent(edm::StreamID const &streamID)
void preGlobalEndLumi(edm::GlobalContext const &globalContext)
void watchPostGlobalEndRun(PostGlobalEndRun::slot_type const &iSlot)
void watchPreBeginJob(PreBeginJob::slot_type const &iSlot)
convenience function for attaching to signal
void postEndRun(edm::GlobalContext const &globalContext)
void watchPreSourceEvent(PreSourceEvent::slot_type const &iSlot)
void preallocate(edm::service::SystemBounds const &bounds)
struct flock data_rw_fulk
unsigned int fuLockPollInterval_
struct flock bu_r_flk
evf::EvFDaqDirector::~EvFDaqDirector ( )

Definition at line 257 of file EvFDaqDirector.cc.

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

258  {
259  if (fulocal_rwlock_fd_!=-1) {
260  unlockFULocal();
261  close(fulocal_rwlock_fd_);
262  }
263 
264  if (fulocal_rwlock_fd2_!=-1) {
265  unlockFULocal2();
266  close(fulocal_rwlock_fd2_);
267  }
268 
269  }

Member Function Documentation

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

Definition at line 72 of file EvFDaqDirector.h.

References run_dir_.

72 {return run_dir_;}
std::string& evf::EvFDaqDirector::buBaseRunDir ( )
inline

Definition at line 73 of file EvFDaqDirector.h.

References bu_run_dir_.

73 {return bu_run_dir_;}
std::string bu_run_dir_
std::string& evf::EvFDaqDirector::buBaseRunOpenDir ( )
inline

Definition at line 74 of file EvFDaqDirector.h.

References bu_run_open_dir_.

74 {return bu_run_open_dir_;}
std::string 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 718 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

718  {
719 
720  if (previousFileSize_ != 0) {
721  if (!fms_) {
722  fms_ = (FastMonitoringService *) (edm::Service<evf::MicroStateService>().operator->());
723  }
725  previousFileSize_ = 0;
726  }
727 
728  //reached limit
729  if (maxLS>=0 && ls > (unsigned int)maxLS) return false;
730 
731  struct stat buf;
732  std::stringstream ss;
733  unsigned int nextIndex = index;
734  nextIndex++;
735 
736  // 1. Check suggested file
737  nextFile = getInputJsonFilePath(ls,index);
738  if (stat(nextFile.c_str(), &buf) == 0) {
739 
740  previousFileSize_ = buf.st_size;
741  fsize = buf.st_size;
742  return true;
743  }
744  // 2. No file -> lumi ended? (and how many?)
745  else {
746  std::string BUEoLSFile = getEoLSFilePathOnBU(ls);
747  bool eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
748  while (eolFound) {
749 
750  // recheck that no raw file appeared in the meantime
751  if (stat(nextFile.c_str(), &buf) == 0) {
752  previousFileSize_ = buf.st_size;
753  fsize = buf.st_size;
754  return true;
755  }
756 
757  int indexFilesInLS = getNFilesFromEoLS(BUEoLSFile);
758  if (indexFilesInLS < 0)
759  //parsing failed
760  return false;
761  else {
762  //check index
763  if ((int)index<indexFilesInLS) {
764  //we have 2 files, and check for 1 failed... retry (2 will never be here)
765  edm::LogError("EvFDaqDirector") << "Potential miss of index file in LS -: " << ls << ". Missing "
766  << nextFile << " because " << indexFilesInLS-1 << " is the highest index expected. Will not update fu.lock file";
767  return false;
768  }
769  }
770  // this lumi ended, check for files
771  ++ls;
772  index = 0;
773 
774  //reached limit
775  if (maxLS>=0 && ls > (unsigned int)maxLS) return false;
776 
777  nextFile = getInputJsonFilePath(ls,0);
778  if (stat(nextFile.c_str(), &buf) == 0) {
779  // a new file was found at new lumisection, index 0
780  previousFileSize_ = buf.st_size;
781  fsize = buf.st_size;
782  return true;
783  }
784  else {
785  //change of policy: we need to cycle through each LS
786  return false;
787  }
788  BUEoLSFile = getEoLSFilePathOnBU(ls);
789  eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
790  }
791  }
792  // no new file found
793  return false;
794  }
void accumulateFileSize(unsigned int lumi, unsigned long fileSize)
def ls
Definition: eostools.py:348
std::string getEoLSFilePathOnBU(const unsigned int ls) const
unsigned long previousFileSize_
int getNFilesFromEoLS(std::string BUEoLSFile)
evf::FastMonitoringService * fms_
std::string getInputJsonFilePath(const unsigned int ls, const unsigned int index) const
void evf::EvFDaqDirector::checkMergeTypePSet ( edm::ProcessContext const &  pc)

Definition at line 1007 of file EvFDaqDirector.cc.

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

Referenced by preBeginJob().

1008  {
1009  if (mergeTypePset_.empty()) return;
1010  if(mergeTypeMap_.size()) return;
1011  edm::ParameterSet const& topPset = edm::getParameterSet(pc.parameterSetID());
1012  if (topPset.existsAs<edm::ParameterSet>(mergeTypePset_,true))
1013  {
1014  const edm::ParameterSet& tsPset(topPset.getParameterSet(mergeTypePset_));
1015  for (std::string pname : tsPset.getParameterNames()) {
1016  std::string streamType = tsPset.getParameter<std::string>(pname);
1017  mergeTypeMap_[pname]=streamType;
1018  }
1019  }
1020  }
bool existsAs(std::string const &parameterName, bool trackiness=true) const
checks if a parameter exists as a given type
Definition: ParameterSet.h:186
ParameterSet const & getParameterSet(ParameterSetID const &id)
std::map< std::string, std::string > mergeTypeMap_
std::string mergeTypePset_
ParameterSet const & getParameterSet(std::string const &) const
void evf::EvFDaqDirector::checkTransferSystemPSet ( edm::ProcessContext const &  pc)

Definition at line 904 of file EvFDaqDirector.cc.

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

Referenced by preBeginJob().

905  {
906  if(transferSystemJson_) return;
907 
908  transferSystemJson_.reset(new Json::Value);
909  edm::ParameterSet const& topPset = edm::getParameterSet(pc.parameterSetID());
910  if (topPset.existsAs<edm::ParameterSet>("transferSystem",true))
911  {
912  const edm::ParameterSet& tsPset(topPset.getParameterSet("transferSystem"));
913 
914  Json::Value destinationsVal(Json::arrayValue);
915  std::vector<std::string> destinations = tsPset.getParameter<std::vector<std::string>>("destinations");
916  for (auto & dest: destinations) destinationsVal.append(dest);
917  (*transferSystemJson_)["destinations"]=destinationsVal;
918 
919  Json::Value modesVal(Json::arrayValue);
920  std::vector<std::string> modes = tsPset.getParameter< std::vector<std::string> >("transferModes");
921  for (auto & mode: modes) modesVal.append(mode);
922  (*transferSystemJson_)["transferModes"]=modesVal;
923 
924  for (auto psKeyItr =tsPset.psetTable().begin();psKeyItr!=tsPset.psetTable().end(); ++ psKeyItr) {
925  if (psKeyItr->first!="destinations" && psKeyItr->first!="transferModes") {
926  const edm::ParameterSet & streamDef = tsPset.getParameterSet(psKeyItr->first);
927  Json::Value streamVal;
928  for (auto & mode : modes) {
929  //validation
930  if (!streamDef.existsAs<std::vector<std::string>>(mode,true))
931  throw cms::Exception("EvFDaqDirector") << " Missing transfer system specification for -:" << psKeyItr->first << " (transferMode " << mode << ")";
932  std::vector<std::string> streamDestinations = streamDef.getParameter<std::vector<std::string>>(mode);
933 
934  Json::Value sDestsValue(Json::arrayValue);
935 
936  if (!streamDestinations.size())
937  throw cms::Exception("EvFDaqDirector") << " Missing transter system destination(s) for -: "<< psKeyItr->first << ", mode:" << mode;
938 
939  for (auto & sdest:streamDestinations) {
940  bool sDestValid=false;
941  sDestsValue.append(sdest);
942  for (auto & dest: destinations) {
943  if (dest==sdest) sDestValid=true;
944  }
945  if (!sDestValid)
946  throw cms::Exception("EvFDaqDirector") << " Invalid transter system destination specified for -: "<< psKeyItr->first << ", mode:" << mode << ", dest:"<<sdest;
947  }
948  streamVal[mode]=sDestsValue;
949  }
950  (*transferSystemJson_)[psKeyItr->first] = streamVal;
951  }
952  }
953  }
954  else {
955  if (requireTSPSet_)
956  throw cms::Exception("EvFDaqDirector") << "transferSystem PSet not found";
957  }
958  }
T getParameter(std::string const &) const
bool existsAs(std::string const &parameterName, bool trackiness=true) const
checks if a parameter exists as a given type
Definition: ParameterSet.h:186
std::shared_ptr< Json::Value > transferSystemJson_
ParameterSet const & getParameterSet(ParameterSetID const &id)
Represents a JSON value.
Definition: value.h:111
ParameterSet const & getParameterSet(std::string const &) const
array value (ordered list)
Definition: value.h:31
void evf::EvFDaqDirector::createProcessingNotificationMaybe ( ) const

Definition at line 1035 of file EvFDaqDirector.cc.

References run_dir_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::checkNextEvent().

1035  {
1036  std::string proc_flag = run_dir_ + "/processing";
1037  int proc_flag_fd = open(proc_flag.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
1038  close(proc_flag_fd);
1039  }
void evf::EvFDaqDirector::createRunOpendirMaybe ( )

Definition at line 876 of file EvFDaqDirector.cc.

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

Referenced by initRun().

876  {
877  // create open dir if not already there
878 
880  if (!boost::filesystem::is_directory(openPath)) {
881  LogDebug("EvFDaqDirector") << "<open> FU dir not found. Creating... -:" << openPath.string();
882  boost::filesystem::create_directories(openPath);
883  }
884  }
#define LogDebug(id)
std::string getRunOpenDirPath() const
bool evf::EvFDaqDirector::emptyLumisectionMode ( ) const
inline

Definition at line 131 of file EvFDaqDirector.h.

References emptyLumisectionMode_.

Referenced by FedRawDataInputSource::checkNextEvent().

131 {return emptyLumisectionMode_;}
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 283 of file EvFDaqDirector.cc.

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

284  {
286  desc.setComment("Service used for file locking arbitration and for propagating information between other EvF components");
287  desc.addUntracked<std::string> ("baseDir", ".")->setComment("Local base directory for run output");
288  desc.addUntracked<std::string> ("buBaseDir", ".")->setComment("BU base ramdisk directory ");
289  desc.addUntracked<unsigned int> ("runNumber",0)->setComment("Run Number in ramdisk to open");
290  desc.addUntracked<bool>("outputAdler32Recheck",false)->setComment("Check Adler32 of per-process output files while micro-merging");
291  desc.addUntracked<bool>("requireTransfersPSet",false)->setComment("Require complete transferSystem PSet in the process configuration");
292  desc.addUntracked<std::string>("selectedTransferMode","")->setComment("Selected transfer mode (choice in Lvl0 propagated as Python parameter");
293  desc.addUntracked<unsigned int>("fuLockPollInterval",2000)->setComment("Lock polling interval in microseconds for the input directory file lock");
294  desc.addUntracked<bool>("emptyLumisectionMode",true)->setComment("Enables writing stream output metadata even when no events are processed in a lumisection");
295  desc.addUntracked<bool>("microMergeDisabled",true)->setComment("Disabled micro-merging by the Output Module, so it is later done by hltd service");
296  desc.addUntracked<std::string>("mergingPset","")->setComment("Name of merging PSet to look for merging type definitions for streams");
297  desc.setAllowAnything();
298  descriptions.add("EvFDaqDirector", desc);
299  }
ParameterDescriptionBase * addUntracked(U const &iLabel, T const &value)
void setAllowAnything()
allow any parameter label/value pairs
void setComment(std::string const &value)
void add(std::string const &label, ParameterSetDescription const &psetDescription)
std::string evf::EvFDaqDirector::findCurrentRunDir ( )
inline

Definition at line 76 of file EvFDaqDirector.h.

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

76 { return dirManager_.findRunDir(run_);}
std::string findRunDir(unsigned int)
Definition: DirManager.cc:37
std::string evf::EvFDaqDirector::getBoLSFilePathOnFU ( const unsigned int  ls) const

Definition at line 458 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::createBoLSFile().

458  {
459  return run_dir_ + "/" + fffnaming::bolsFileName(run_,ls);
460  }
std::string bolsFileName(const unsigned int run, const unsigned int ls)
def ls
Definition: eostools.py:348
std::string evf::EvFDaqDirector::getDatFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 394 of file EvFDaqDirector.cc.

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

394  {
396  }
def ls
Definition: eostools.py:348
std::string streamerDataFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getEoLSFilePathOnBU ( const unsigned int  ls) const

Definition at line 450 of file EvFDaqDirector.cc.

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

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

450  {
451  return bu_run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
452  }
def ls
Definition: eostools.py:348
std::string bu_run_dir_
std::string eolsFileName(const unsigned int run, const unsigned int ls)
std::string evf::EvFDaqDirector::getEoLSFilePathOnFU ( const unsigned int  ls) const

Definition at line 454 of file EvFDaqDirector.cc.

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

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

454  {
455  return run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
456  }
def ls
Definition: eostools.py:348
std::string eolsFileName(const unsigned int run, const unsigned int ls)
std::string evf::EvFDaqDirector::getEoRFilePath ( ) const

Definition at line 462 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

462  {
463  return bu_run_dir_ + "/" + fffnaming::eorFileName(run_);
464  }
std::string eorFileName(const unsigned int run)
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getEoRFilePathOnFU ( ) const

Definition at line 467 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::checkNextEvent().

467  {
468  return run_dir_ + "/" + fffnaming::eorFileName(run_);
469  }
std::string eorFileName(const unsigned int run)
std::string evf::EvFDaqDirector::getInitFilePath ( std::string const &  stream) const

Definition at line 422 of file EvFDaqDirector.cc.

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

Referenced by DQMFileSaver::globalBeginRun().

422  {
423  return run_dir_ + "/" + fffnaming::initFileNameWithPid(run_,0,stream);
424  }
std::string initFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getInputJsonFilePath ( const unsigned int  ls,
const unsigned int  index 
) const

Definition at line 377 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

377  {
379  }
def ls
Definition: eostools.py:348
std::string inputJsonFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getMergedDatChecksumFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 414 of file EvFDaqDirector.cc.

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

414  {
416  }
std::string streamerDataChecksumFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
def ls
Definition: eostools.py:348
std::string evf::EvFDaqDirector::getMergedDatFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 410 of file EvFDaqDirector.cc.

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

Referenced by maybeCreateAndLockFileHeadForStream().

410  {
412  }
def ls
Definition: eostools.py:348
std::string streamerDataFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string evf::EvFDaqDirector::getMergedProtocolBufferHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 434 of file EvFDaqDirector.cc.

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

434  {
436  }
std::string protocolBufferHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
def ls
Definition: eostools.py:348
std::string evf::EvFDaqDirector::getMergedRootHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 446 of file EvFDaqDirector.cc.

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

446  {
448  }
def ls
Definition: eostools.py:348
std::string rootHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
int evf::EvFDaqDirector::getNFilesFromEoLS ( std::string  BUEoLSFile)
private

Definition at line 657 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

657  {
658 
659  boost::filesystem::ifstream ij(BUEoLSFile);
660  Json::Value deserializeRoot;
662 
663  if (!reader.parse(ij, deserializeRoot)) {
664  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << BUEoLSFile;
665  return -1;
666  }
667 
669  DataPoint dp;
670  dp.deserialize(deserializeRoot);
671 
672  //read definition
673  if (readEolsDefinition_) {
674  //std::string def = boost::algorithm::trim(dp.getDefinition());
676  if (!def.size()) readEolsDefinition_=false;
677  while (def.size()) {
678  std::string fullpath;
679  if (def.find('/')==0)
680  fullpath = def;
681  else
682  fullpath = bu_run_dir_+'/'+def;
683  struct stat buf;
684  if (stat(fullpath.c_str(), &buf) == 0) {
685  DataPointDefinition eolsDpd;
686  std::string defLabel = "legend";
687  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
688  if (eolsDpd.getNames().size()==0) {
689  //try with "data" label if "legend" format is not used
690  eolsDpd = DataPointDefinition();
691  defLabel="data";
692  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
693  }
694  for (unsigned int i=0;i<eolsDpd.getNames().size();i++)
695  if (eolsDpd.getNames().at(i)=="NFiles")
697  readEolsDefinition_=false;
698  break;
699  }
700  //check if we can still find definition
701  if (def.size()<=1 || def.find('/')==std::string::npos) {
702  readEolsDefinition_=false;
703  break;
704  }
705  def = def.substr(def.find('/')+1);
706  }
707  }
708 
709  if (dp.getData().size()>eolsNFilesIndex_)
710  data = dp.getData()[eolsNFilesIndex_];
711  else {
712  edm::LogError("EvFDaqDirector") << " error reading number of files from BU JSON -: " << BUEoLSFile;
713  return -1;
714  }
715  return boost::lexical_cast<int>(data);
716  }
std::vector< std::string > & getData()
Definition: DataPoint.h:58
int i
Definition: DBlmapReader.cc:9
bool parse(const std::string &document, Value &root, bool collectComments=true)
Read a Value from a JSON document.
Represents a JSON value.
Definition: value.h:111
virtual void deserialize(Json::Value &root)
Definition: DataPoint.cc:56
unsigned int eolsNFilesIndex_
auto dp
Definition: deltaR.h:22
std::string & getDefinition()
Definition: DataPoint.h:59
std::string bu_run_dir_
Unserialize a JSON document into a Value.
Definition: reader.h:16
std::vector< std::string > const & getNames()
JetCorrectorParameters::Definitions def
Definition: classes.h:6
std::string evf::EvFDaqDirector::getOpenDatFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 398 of file EvFDaqDirector.cc.

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

398  {
399  return run_dir_ + "/open/" + fffnaming::streamerDataFileNameWithPid(run_,ls,stream);
400  }
def ls
Definition: eostools.py:348
std::string streamerDataFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getOpenInitFilePath ( std::string const &  stream) const

Definition at line 418 of file EvFDaqDirector.cc.

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

418  {
419  return run_dir_ + "/open/" + fffnaming::initFileNameWithPid(run_,0,stream);
420  }
std::string initFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getOpenInputJsonFilePath ( const unsigned int  ls,
const unsigned int  index 
) const

Definition at line 390 of file EvFDaqDirector.cc.

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

390  {
392  }
def ls
Definition: eostools.py:348
std::string inputJsonFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getOpenOutputJsonFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 402 of file EvFDaqDirector.cc.

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

402  {
403  return run_dir_ + "/open/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
404  }
def ls
Definition: eostools.py:348
std::string streamerJsonFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getOpenProtocolBufferHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 426 of file EvFDaqDirector.cc.

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

426  {
428  }
def ls
Definition: eostools.py:348
std::string protocolBufferHistogramFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getOpenRawFilePath ( const unsigned int  ls,
const unsigned int  index 
) const

Definition at line 386 of file EvFDaqDirector.cc.

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

386  {
387  return bu_run_dir_ + "/open/" + fffnaming::inputRawFileName(run_,ls,index);
388  }
std::string inputRawFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
def ls
Definition: eostools.py:348
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getOpenRootHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 438 of file EvFDaqDirector.cc.

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

438  {
439  return run_dir_ + "/open/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
440  }
def ls
Definition: eostools.py:348
std::string rootHistogramFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getOutputJsonFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 406 of file EvFDaqDirector.cc.

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

406  {
408  }
def ls
Definition: eostools.py:348
std::string streamerJsonFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getProtocolBufferHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 430 of file EvFDaqDirector.cc.

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

430  {
432  }
def ls
Definition: eostools.py:348
std::string protocolBufferHistogramFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string evf::EvFDaqDirector::getRawFilePath ( const unsigned int  ls,
const unsigned int  index 
) const

Definition at line 382 of file EvFDaqDirector.cc.

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

Referenced by removeFile().

382  {
384  }
std::string inputRawFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
def ls
Definition: eostools.py:348
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getRootHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 442 of file EvFDaqDirector.cc.

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

442  {
444  }
def ls
Definition: eostools.py:348
std::string rootHistogramFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
unsigned int evf::EvFDaqDirector::getRunNumber ( ) const
inline

Definition at line 107 of file EvFDaqDirector.h.

References run_.

107 { return run_; }
std::string evf::EvFDaqDirector::getRunOpenDirPath ( ) const
inline

Definition at line 100 of file EvFDaqDirector.h.

References run_dir_.

Referenced by createRunOpendirMaybe(), and initRun().

100 {return run_dir_ +"/open";}
std::string evf::EvFDaqDirector::getStreamDestinations ( std::string const &  stream) const

Definition at line 960 of file EvFDaqDirector.cc.

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

961  {
962  std::string streamRequestName;
963  if (transferSystemJson_->isMember(stream.c_str()))
964  streamRequestName = stream;
965  else {
966  std::stringstream msg;
967  msg << "Transfer system mode definitions missing for -: " << stream;
968  if (requireTSPSet_)
969  throw cms::Exception("EvFDaqDirector") << msg.str();
970  else {
971  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
972  return std::string("Failsafe");
973  }
974  }
975  //return empty if strict check parameter is not on
977  edm::LogWarning("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter."
978  << "Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
979  return std::string("Failsafe");
980  }
982  throw cms::Exception("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter.";
983  }
984  //check if stream has properly listed transfer stream
985  if (!transferSystemJson_->get(streamRequestName, "").isMember(selectedTransferMode_.c_str()))
986  {
987  std::stringstream msg;
988  msg << "Selected transfer mode " << selectedTransferMode_ << " is not specified for stream " << streamRequestName;
989  if (requireTSPSet_)
990  throw cms::Exception("EvFDaqDirector") << msg.str();
991  else
992  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
993  return std::string("Failsafe");
994  }
995  Json::Value destsVec = transferSystemJson_->get(streamRequestName, "").get(selectedTransferMode_,"");
996 
997  //flatten string json::Array into CSV std::string
999  for (Json::Value::iterator it = destsVec.begin(); it!=destsVec.end(); it++)
1000  {
1001  if (ret!="") ret +=",";
1002  ret+=(*it).asString();
1003  }
1004  return ret;
1005  }
tuple ret
prodAgent to be discontinued
const_iterator begin() const
std::shared_ptr< Json::Value > transferSystemJson_
Represents a JSON value.
Definition: value.h:111
std::string selectedTransferMode_
const_iterator end() const
Iterator for object and array value.
Definition: value.h:1007
std::vector<int>* evf::EvFDaqDirector::getStreamFileTracker ( )
inline

Definition at line 114 of file EvFDaqDirector.h.

References streamFileTracker_.

Referenced by FedRawDataInputSource::getNextEvent().

114 {return &streamFileTracker_;}
std::vector< int > streamFileTracker_
std::string evf::EvFDaqDirector::getStreamMergeType ( std::string const &  stream,
MergeType  defaultType 
)

Definition at line 1022 of file EvFDaqDirector.cc.

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

1023  {
1024  auto mergeTypeItr = mergeTypeMap_.find(stream.c_str());
1025  if (mergeTypeItr == mergeTypeMap_.end()) {
1026  edm::LogInfo("EvFDaqDirector") << " No merging type specified for stream " << stream << ". Using default value";
1027  assert(defaultType<MergeTypeNames_.size());
1028  std::string defaultName = MergeTypeNames_[defaultType];
1029  mergeTypeMap_[stream] = defaultName;
1030  return defaultName;
1031  }
1032  return mergeTypeItr->second;
1033  }
assert(m_qm.get())
std::map< std::string, std::string > mergeTypeMap_
static const std::vector< std::string > MergeTypeNames_
std::string evf::EvFDaqDirector::initFileName ( std::string const &  stream) const
private
void evf::EvFDaqDirector::initRun ( void  )

Definition at line 130 of file EvFDaqDirector.cc.

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

Referenced by preallocate().

131  {
132  std::stringstream ss;
133  ss << "run" << std::setfill('0') << std::setw(6) << run_;
134  run_string_ = ss.str();
136 
137  // check if base dir exists or create it accordingly
138  int retval = mkdir(base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
139  if (retval != 0 && errno != EEXIST) {
140  throw cms::Exception("DaqDirector") << " Error checking for base dir -: "
141  << base_dir_ << " mkdir error:" << strerror(errno);
142  }
143 
144  //create run dir in base dir
145  umask(0);
146  retval = mkdir(run_dir_.c_str(),
147  S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
148  if (retval != 0 && errno != EEXIST) {
149  throw cms::Exception("DaqDirector") << " Error creating run dir -: "
150  << run_dir_ << " mkdir error:" << strerror(errno);
151  }
152 
153  //create fu-local.lock in run open dir
154  if (!directorBu_) {
155 
157  std::string fulocal_lock_ = getRunOpenDirPath() +"/fu-local.lock";
158  fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
159  if (fulocal_rwlock_fd_==-1)
160  throw cms::Exception("DaqDirector") << " Error creating/opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
161  chmod(fulocal_lock_.c_str(),0777);
162  fsync(fulocal_rwlock_fd_);
163  //open second fd for another input source thread
164  fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
165  if (fulocal_rwlock_fd2_==-1)
166  throw cms::Exception("DaqDirector") << " Error opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
167  }
168 
169  //bu_run_dir: for FU, for which the base dir is local and the BU is remote, it is expected to be there
170  //for BU, it is created at this point
171  if (directorBu_)
172  {
174  std::string bulockfile = bu_run_dir_ + "/bu.lock";
175  std::string fulockfile = bu_run_dir_ + "/fu.lock";
176 
177  //make or find bu run dir
178  retval = mkdir(bu_run_dir_.c_str(),
179  S_IRWXU | S_IRWXG | S_IRWXO);
180  if (retval != 0 && errno != EEXIST) {
181  throw cms::Exception("DaqDirector")
182  << " Error creating bu run dir -: " << bu_run_dir_
183  << " mkdir error:" << strerror(errno) << "\n";
184  }
185  bu_run_open_dir_ = bu_run_dir_ + "/open";
186  retval = mkdir(bu_run_open_dir_.c_str(),
187  S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
188  if (retval != 0 && errno != EEXIST) {
189  throw cms::Exception("DaqDirector") << " Error creating bu run open dir -: "
190  << bu_run_open_dir_ << " mkdir error:" << strerror(errno)
191  << "\n";
192  }
193 
194  // the BU director does not need to know about the fu lock
195  bu_writelock_fd_ = open(bulockfile.c_str(),
196  O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
197  if (bu_writelock_fd_ == -1)
198  edm::LogWarning("EvFDaqDirector") << "problem with creating filedesc for buwritelock -: "
199  << strerror(errno);
200  else
201  edm::LogInfo("EvFDaqDirector") << "creating filedesc for buwritelock -: "
202  << bu_writelock_fd_;
203  bu_w_lock_stream = fdopen(bu_writelock_fd_, "w");
204  if (bu_w_lock_stream == 0)
205  edm::LogWarning("EvFDaqDirector")<< "Error creating write lock stream -: " << strerror(errno);
206 
207  // BU INITIALIZES LOCK FILE
208  // FU LOCK FILE OPEN
209  openFULockfileStream(fulockfile, true);
211  fflush(fu_rw_lock_stream);
212  close(fu_readwritelock_fd_);
213 
214  if (hltSourceDirectory_.size())
215  {
216  struct stat buf;
217  if (stat(hltSourceDirectory_.c_str(),&buf)==0) {
218  std::string hltdir=bu_run_dir_+"/hlt";
219  std::string tmphltdir=bu_run_open_dir_+"/hlt";
220  retval = mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
221  if (retval != 0 && errno != EEXIST)
222  throw cms::Exception("DaqDirector")
223  << " Error creating bu run dir -: " << hltdir
224  << " mkdir error:" << strerror(errno) << "\n";
225 
226  boost::filesystem::copy_file(hltSourceDirectory_+"/HltConfig.py",tmphltdir+"/HltConfig.py");
227 
228  boost::filesystem::copy_file(hltSourceDirectory_+"/fffParameters.jsn",tmphltdir+"/fffParameters.jsn");
229 
230  boost::filesystem::rename(tmphltdir,hltdir);
231  }
232  else
233  throw cms::Exception("DaqDirector") << " Error looking for HLT configuration -: " << hltSourceDirectory_;
234  }
235  //else{}//no configuration specified
236  }
237  else
238  {
239  // for FU, check if bu base dir exists
240 
241  retval = mkdir(bu_base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
242  if (retval != 0 && errno != EEXIST) {
243  throw cms::Exception("DaqDirector") << " Error checking for bu base dir -: "
244  << bu_base_dir_ << " mkdir error:" << strerror(errno) << "\n";
245  }
246 
248  std::string fulockfile = bu_run_dir_ + "/fu.lock";
249  openFULockfileStream(fulockfile, false);
250  }
251 
252  pthread_mutex_init(&init_lock_,NULL);
253 
254  stopFilePath_ = run_dir_+"/CMSSW_STOP";
255  }
std::string run_string_
void openFULockfileStream(std::string &fuLockFilePath, bool create)
pthread_mutex_t init_lock_
#define NULL
Definition: scimark2.h:8
def chmod
Definition: eostools.py:293
std::string hltSourceDirectory_
std::string stopFilePath_
std::string bu_base_dir_
std::string bu_run_dir_
def mkdir
Definition: eostools.py:250
std::string getRunOpenDirPath() const
std::string bu_run_open_dir_
std::string evf::EvFDaqDirector::inputFileNameStem ( const unsigned int  ls,
const unsigned int  index 
) const
private
bool evf::EvFDaqDirector::isSingleStreamThread ( )
inline

Definition at line 115 of file EvFDaqDirector.h.

References nStreams_, and nThreads_.

Referenced by FedRawDataInputSource::getNextEvent().

115 {return nStreams_==1 && nThreads_==1;}
unsigned int nThreads_
unsigned int nStreams_
void evf::EvFDaqDirector::lockFULocal ( )

Definition at line 854 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by updateFuLock().

854  {
855  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_flk);
856  flock(fulocal_rwlock_fd_,LOCK_SH);
857  }
void evf::EvFDaqDirector::lockFULocal2 ( )

Definition at line 865 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

865  {
866  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_flk2);
867  flock(fulocal_rwlock_fd2_,LOCK_EX);
868  }
void evf::EvFDaqDirector::lockInitLock ( )

Definition at line 846 of file EvFDaqDirector.cc.

References init_lock_.

846  {
847  pthread_mutex_lock(&init_lock_);
848  }
pthread_mutex_t init_lock_
FILE * evf::EvFDaqDirector::maybeCreateAndLockFileHeadForStream ( unsigned int  ls,
std::string &  stream 
)

Definition at line 826 of file EvFDaqDirector.cc.

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

826  {
827  data_rw_stream = fopen(getMergedDatFilePath(ls,stream).c_str(), "a"); //open stream for appending
829  if (data_readwrite_fd_ == -1)
830  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for datamerge "
831  << strerror(errno);
832  else
833  LogDebug("EvFDaqDirector") << "creating filedesc for datamerge -: "
835  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_flk);
836 
837  return data_rw_stream;
838  }
#define LogDebug(id)
def ls
Definition: eostools.py:348
struct flock data_rw_flk
std::string getMergedDatFilePath(const unsigned int ls, std::string const &stream) const
std::string evf::EvFDaqDirector::mergedFileNameStem ( const unsigned int  ls,
std::string const &  stream 
) const
private
bool evf::EvFDaqDirector::microMergeDisabled ( ) const
inline

Definition at line 132 of file EvFDaqDirector.h.

References microMergeDisabled_.

132 {return microMergeDisabled_;}
void evf::EvFDaqDirector::openFULockfileStream ( std::string &  fuLockFilePath,
bool  create 
)
private

Definition at line 807 of file EvFDaqDirector.cc.

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

Referenced by initRun().

807  {
808  if (create) {
809  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
810  S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
811  chmod(fulockfile.c_str(),0766);
812  } else {
813  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
814  }
815  if (fu_readwritelock_fd_ == -1)
816  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
817  << " create:" << create << " error:" << strerror(errno);
818  else
819  LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: "
821 
822  fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
823  }
#define LogDebug(id)
def chmod
Definition: eostools.py:293
SurfaceDeformation * create(int type, const std::vector< double > &params)
bool evf::EvFDaqDirector::outputAdler32Recheck ( ) const
inline

Definition at line 101 of file EvFDaqDirector.h.

References outputAdler32Recheck_.

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

References DTTTrigCorrFirst::run, and run_.

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

Definition at line 319 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

319  {
320  close(bu_readlock_fd_);
321  close(bu_writelock_fd_);
322  if (directorBu_) {
323  std::string filename = bu_run_dir_ + "/bu.lock";
324  removeFile(filename);
325  }
326  }
void removeFile(unsigned int ls, unsigned int index)
std::string bu_run_dir_
tuple filename
Definition: lut2db_cfg.py:20
void evf::EvFDaqDirector::preallocate ( edm::service::SystemBounds const &  bounds)

Definition at line 272 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

272  {
273 
274  initRun();
275 
276  for (unsigned int i=0;i<bounds.maxNumberOfStreams();i++){
277  streamFileTracker_.push_back(-1);
278  }
279  nThreads_=bounds.maxNumberOfStreams();
280  nStreams_=bounds.maxNumberOfThreads();
281  }
unsigned int nThreads_
int i
Definition: DBlmapReader.cc:9
std::vector< int > streamFileTracker_
unsigned int nStreams_
void evf::EvFDaqDirector::preBeginJob ( edm::PathsAndConsumesOfModulesBase const &  ,
edm::ProcessContext const &  pc 
)

Definition at line 301 of file EvFDaqDirector.cc.

References checkMergeTypePSet(), and checkTransferSystemPSet().

Referenced by EvFDaqDirector().

302  {
304  checkMergeTypePSet(pc);
305  }
void checkMergeTypePSet(edm::ProcessContext const &pc)
void checkTransferSystemPSet(edm::ProcessContext const &pc)
void evf::EvFDaqDirector::preBeginRun ( edm::GlobalContext const &  globalContext)

Definition at line 307 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

307  {
308 
309  //assert(run_ == id.run());
310 
311  // check if the requested run is the latest one - issue a warning if it isn't
313  edm::LogWarning("EvFDaqDirector") << "WARNING - checking run dir -: "
314  << run_dir_ << ". This is not the highest run "
316  }
317  }
std::string findHighestRunDir()
Definition: DirManager.cc:20
void evf::EvFDaqDirector::preGlobalEndLumi ( edm::GlobalContext const &  globalContext)

Definition at line 328 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

329  {
330  //delete all files belonging to just closed lumi
331  unsigned int ls = globalContext.luminosityBlockID().luminosityBlock();
333  edm::LogWarning("EvFDaqDirector") << " Handles to check for files to delete were not set by the input source...";
334  return;
335  }
336 
337  std::unique_lock<std::mutex> lkw(*fileDeleteLockPtr_);
338  auto it = filesToDeletePtr_->begin();
339  while (it!=filesToDeletePtr_->end()) {
340  if (it->second->lumi_ == ls) {
341  const boost::filesystem::path filePath(it->second->fileName_);
342  LogDebug("EvFDaqDirector") << "Deleting input file -:" << it->second->fileName_;
343  try {
344  //rarely this fails but file gets deleted
345  boost::filesystem::remove(filePath);
346  }
347  catch (const boost::filesystem::filesystem_error& ex)
348  {
349  edm::LogError("EvFDaqDirector") << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() << ". Trying again.";
350  usleep(10000);
351  try {
352  boost::filesystem::remove(filePath);
353  }
354  catch (const boost::filesystem::filesystem_error&) {/*file gets deleted first time but exception is still thrown*/}
355  }
356  catch (std::exception& ex)
357  {
358  edm::LogError("EvFDaqDirector") << " - deleteFile std::exception CAUGHT -: " << ex.what() << ". Trying again.";
359  usleep(10000);
360  try {
361  boost::filesystem::remove(filePath);
362  } catch (std::exception&) {/*file gets deleted first time but exception is still thrown*/}
363  }
364 
365  delete it->second;
366  it = filesToDeletePtr_->erase(it);
367  }
368  else it++;
369  }
370  }
#define LogDebug(id)
std::list< std::pair< int, InputFile * > > * filesToDeletePtr_
def ls
Definition: eostools.py:348
std::mutex * fileDeleteLockPtr_
void evf::EvFDaqDirector::preSourceEvent ( edm::StreamID const &  streamID)
inline

Definition at line 372 of file EvFDaqDirector.cc.

References currentFileIndex_, and streamFileTracker_.

Referenced by EvFDaqDirector().

372  {
374  }
std::vector< int > streamFileTracker_
int evf::EvFDaqDirector::readLastLSEntry ( std::string const &  file)

Definition at line 887 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

887  {
888 
889  boost::filesystem::ifstream ij(file);
890  Json::Value deserializeRoot;
892 
893  if (!reader.parse(ij, deserializeRoot)) {
894  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << file;
895  return -1;
896  }
897 
898  int ret = deserializeRoot.get("lastLS","").asInt();
899  return ret;
900 
901  }
Value get(UInt index, const Value &defaultValue) const
tuple ret
prodAgent to be discontinued
bool parse(const std::string &document, Value &root, bool collectComments=true)
Read a Value from a JSON document.
Represents a JSON value.
Definition: value.h:111
Int asInt() const
Unserialize a JSON document into a Value.
Definition: reader.h:16
void evf::EvFDaqDirector::removeFile ( unsigned int  ls,
unsigned int  index 
)

Definition at line 478 of file EvFDaqDirector.cc.

References getRawFilePath().

Referenced by postEndRun().

478  {
480  }
def ls
Definition: eostools.py:348
std::string getRawFilePath(const unsigned int ls, const unsigned int index) const
void removeFile(unsigned int ls, unsigned int index)
void evf::EvFDaqDirector::removeFile ( std::string  filename)

Definition at line 471 of file EvFDaqDirector.cc.

471  {
472  int retval = remove(filename.c_str());
473  if (retval != 0)
474  edm::LogError("EvFDaqDirector") << "Could not remove used file -: " << filename << ". error = "
475  << strerror(errno);
476  }
tuple filename
Definition: lut2db_cfg.py:20
void evf::EvFDaqDirector::setDeleteTracking ( std::mutex fileDeleteLock,
std::list< std::pair< int, InputFile * >> *  filesToDelete 
)
inline

Definition at line 123 of file EvFDaqDirector.h.

References fileDeleteLockPtr_, and filesToDeletePtr_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

123  {
124  fileDeleteLockPtr_=fileDeleteLock;
125  filesToDeletePtr_ = filesToDelete;
126  }
std::list< std::pair< int, InputFile * > > * filesToDeletePtr_
std::mutex * fileDeleteLockPtr_
void evf::EvFDaqDirector::setFMS ( evf::FastMonitoringService fms)
inline

Definition at line 112 of file EvFDaqDirector.h.

References fms_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

112 {fms_=fms;}
evf::FastMonitoringService * fms_
void evf::EvFDaqDirector::tryInitializeFuLockFile ( )

Definition at line 796 of file EvFDaqDirector.cc.

References fu_rw_lock_stream, and getHLTprescales::readIndex().

Referenced by initRun().

796  {
797  if (fu_rw_lock_stream == 0)
798  edm::LogError("EvFDaqDirector") << "Error creating fu read/write lock stream "
799  << strerror(errno);
800  else {
801  edm::LogInfo("EvFDaqDirector") << "Initializing FU LOCK FILE";
802  unsigned int readLs = 1, readIndex = 0;
803  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
804  }
805  }
void evf::EvFDaqDirector::unlockAndCloseMergeStream ( )

Definition at line 840 of file EvFDaqDirector.cc.

References data_readwrite_fd_, data_rw_fulk, and data_rw_stream.

840  {
841  fflush(data_rw_stream);
842  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_fulk);
843  fclose(data_rw_stream);
844  }
struct flock data_rw_fulk
void evf::EvFDaqDirector::unlockFULocal ( )

Definition at line 859 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

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

859  {
860  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_fulk);
861  flock(fulocal_rwlock_fd_,LOCK_UN);
862  }
void evf::EvFDaqDirector::unlockFULocal2 ( )

Definition at line 870 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

870  {
871  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_fulk2);
872  flock(fulocal_rwlock_fd2_,LOCK_UN);
873  }
void evf::EvFDaqDirector::unlockInitLock ( )

Definition at line 850 of file EvFDaqDirector.cc.

References init_lock_.

850  {
851  pthread_mutex_unlock(&init_lock_);
852  }
pthread_mutex_t init_lock_
void evf::EvFDaqDirector::updateFileIndex ( int const &  fileIndex)
inline

Definition at line 113 of file EvFDaqDirector.h.

References currentFileIndex_.

Referenced by FedRawDataInputSource::getNextEvent().

113 {currentFileIndex_=fileIndex;}
EvFDaqDirector::FileStatus evf::EvFDaqDirector::updateFuLock ( unsigned int &  ls,
std::string &  nextFile,
uint32_t &  fsize,
uint64_t &  lockWaitTime 
)

Definition at line 482 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::readSupervisor().

482  {
483  EvFDaqDirector::FileStatus fileStatus = noFile;
484 
485  int retval = -1;
486  int lock_attempts = 0;
487 
488  struct stat buf;
489  int stopFileLS = -1;
490  if (stat(stopFilePath_.c_str(),&buf)==0) {
491  stopFileLS = readLastLSEntry(stopFilePath_);
492  if (!stop_ls_override_) {
493  //if lumisection is higher than in stop file, should quit at next from current
494  if (stopFileLS>=0 && (int)ls>=stopFileLS) stopFileLS = stop_ls_override_ = ls;
495  }
496  else stopFileLS = stop_ls_override_;
497  edm::LogWarning("EvFDaqDirector") << "Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
498  //return runEnded;
499  }
500  else //if file was removed before reaching stop condition, reset this
501  stop_ls_override_ = 0;
502 
503  timeval ts_lockbegin;
504  gettimeofday(&ts_lockbegin,0);
505 
506  while (retval==-1) {
507  retval = fcntl(fu_readwritelock_fd_, F_SETLK, &fu_rw_flk);
508  if (retval==-1) usleep(fuLockPollInterval_);
509  else continue;
510 
511  lock_attempts+=fuLockPollInterval_;
512  if (lock_attempts>5000000 || errno==116) {
513  if (errno==116)
514  edm::LogWarning("EvFDaqDirector") << "Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
515  else
516  edm::LogWarning("EvFDaqDirector") << "Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno "
517  << errno <<":"<< strerror(errno) << std::endl;
518 
519 
520  if (stat(getEoLSFilePathOnFU(ls).c_str(),&buf)==0) {
521  edm::LogWarning("EvFDaqDirector") << "Detected local EoLS for lumisection "<< ls ;
522  ls++;
523  return noFile;
524  }
525 
526  if (stat(bu_run_dir_.c_str(), &buf)!=0) return runEnded;
527  if (stat((bu_run_dir_+"/fu.lock").c_str(), &buf)!=0) return runEnded;
528  lock_attempts=0;
529  }
530  }
531 
532  timeval ts_lockend;
533  gettimeofday(&ts_lockend,0);
534  long deltat = (ts_lockend.tv_usec-ts_lockbegin.tv_usec) + (ts_lockend.tv_sec-ts_lockbegin.tv_sec)*1000000;
535  if (deltat>0.) lockWaitTime=deltat;
536 
537 
538 
539  if(retval!=0) return fileStatus;
540 
541 #ifdef DEBUG
542  timeval ts_lockend;
543  gettimeofday(&ts_lockend,0);
544 #endif
545 
546  // if the stream is readable
547  if (fu_rw_lock_stream != 0) {
548  unsigned int readLs, readIndex;
549  int check = 0;
550  // rewind the stream
551  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
552  // if rewinded ok
553  if (check == 0) {
554  // read its' values
555  fscanf(fu_rw_lock_stream, "%u %u", &readLs, &readIndex);
556  edm::LogInfo("EvFDaqDirector") << "Read fu.lock file file -: " << readLs << ":" << readIndex;
557 
558  unsigned int currentLs = readLs;
559  bool bumpedOk = false;
560  //if next lumisection in a lock file is not +1 wrt. source, cycle through the next empty one, unless initial lumi not yet set
561  //no lock file write in this case
562  if (ls && ls+1 < currentLs) ls++;
563  else {
564  // try to bump (look for new index or EoLS file)
565  bumpedOk = bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
566  //avoid 2 lumisections jump
567  if (ls && readLs>currentLs && currentLs > ls) {
568  ls++;
569  readLs=currentLs=ls;
570  readIndex=0;
571  bumpedOk=false;
572  //no write to lock file
573  }
574  else {
575  if (ls==0 && readLs>currentLs) {
576  //make sure to intialize always with LS found in the lock file, with possibility of grabbing index file immediately
577  //in this case there is no new file in the same LS
578  readLs=currentLs;
579  readIndex=0;
580  bumpedOk=false;
581  //no write to lock file
582  }
583  //update return LS value
584  ls = readLs;
585  }
586  }
587  if (bumpedOk) {
588  // there is a new index file to grab, lock file needs to be updated
589  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
590  if (check == 0) {
591  ftruncate(fu_readwritelock_fd_, 0);
592  // write next index in the file, which is the file the next process should take
593  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex + 1);
594  fflush(fu_rw_lock_stream);
595  fsync(fu_readwritelock_fd_);
596  fileStatus = newFile;
597  LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":" << readIndex + 1;
598  }
599  else {
600  throw cms::Exception("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error " << strerror(errno);
601  }
602  }
603  else if (currentLs < readLs) {
604  //there is no new file in next LS (yet), but lock file can be updated to the next LS
605  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
606  if (check == 0) {
607  ftruncate(fu_readwritelock_fd_, 0);
608  // in this case LS was bumped, but no new file. Thus readIndex is 0 (set by bumpFile)
609  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
610  fflush(fu_rw_lock_stream);
611  fsync(fu_readwritelock_fd_);
612  LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":" << readIndex;
613  }
614  else {
615  throw cms::Exception("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error " << strerror(errno);
616  }
617  }
618  } else {
619  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for reading failed with error " << strerror(errno);
620  }
621  } else {
622  edm::LogError("EvFDaqDirector") << "fu read/write lock stream is invalid " << strerror(errno);
623  }
624 
625 #ifdef DEBUG
626  timeval ts_preunlock;
627  gettimeofday(&ts_preunlock,0);
628  int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
629  double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
630 #endif
631 
632  //if new json is present, lock file which FedRawDataInputSource will later unlock
633  if (fileStatus==newFile) lockFULocal();
634 
635  //release lock at this point
636  int retvalu=-1;
637  retvalu=fcntl(fu_readwritelock_fd_, F_SETLKW, &fu_rw_fulk);
638  if (retvalu==-1) edm::LogError("EvFDaqDirector") << "Error unlocking the fu.lock " << strerror(errno);
639 
640 #ifdef DEBUG
641  edm::LogDebug("EvFDaqDirector") << "Waited during lock -: " << locked_period << " seconds";
642 #endif
643 
644  if ( fileStatus == noFile ) {
645  struct stat buf;
646  //edm::LogInfo("EvFDaqDirector") << " looking for EoR file: " << getEoRFilePath().c_str();
647  if ( stat(getEoRFilePath().c_str(), &buf) == 0 || stat(bu_run_dir_.c_str(), &buf)!=0)
648  fileStatus = runEnded;
649  if (stopFileLS>=0 && (int)ls > stopFileLS) {
650  edm::LogInfo("EvFDaqDirector") << "Reached maximum lumisection set by hltd";
651  fileStatus = runEnded;
652  }
653  }
654  return fileStatus;
655  }
#define LogDebug(id)
struct flock fu_rw_flk
def ls
Definition: eostools.py:348
std::string getEoRFilePath() const
struct flock fu_rw_fulk
bool check(const std::string &)
std::string stopFilePath_
unsigned int stop_ls_override_
int readLastLSEntry(std::string const &file)
std::string bu_run_dir_
std::string getEoLSFilePathOnFU(const unsigned int ls) const
bool bumpFile(unsigned int &ls, unsigned int &index, std::string &nextFile, uint32_t &fsize, int maxLS)
unsigned int fuLockPollInterval_

Member Data Documentation

std::string evf::EvFDaqDirector::base_dir_
private

Definition at line 148 of file EvFDaqDirector.h.

Referenced by initRun().

std::string evf::EvFDaqDirector::bu_base_dir_
private

Definition at line 149 of file EvFDaqDirector.h.

Referenced by initRun().

struct flock evf::EvFDaqDirector::bu_r_flk
private

Definition at line 186 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_r_fulk
private

Definition at line 188 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_r_lock_stream
private

Definition at line 175 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_readlock_fd_
private

Definition at line 167 of file EvFDaqDirector.h.

Referenced by postEndRun().

std::string evf::EvFDaqDirector::bu_run_dir_
private
std::string evf::EvFDaqDirector::bu_run_open_dir_
private

Definition at line 165 of file EvFDaqDirector.h.

Referenced by buBaseRunOpenDir(), and initRun().

FILE* evf::EvFDaqDirector::bu_t_monitor_stream
private

Definition at line 178 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_flk
private

Definition at line 185 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_fulk
private

Definition at line 187 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_w_lock_stream
private

Definition at line 174 of file EvFDaqDirector.h.

Referenced by initRun().

FILE* evf::EvFDaqDirector::bu_w_monitor_stream
private

Definition at line 177 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_writelock_fd_
private

Definition at line 168 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

int evf::EvFDaqDirector::currentFileIndex_ = -1
private

Definition at line 200 of file EvFDaqDirector.h.

Referenced by preSourceEvent(), and updateFileIndex().

int evf::EvFDaqDirector::data_readwrite_fd_
private
struct flock evf::EvFDaqDirector::data_rw_flk
private

Definition at line 191 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream().

struct flock evf::EvFDaqDirector::data_rw_fulk
private

Definition at line 192 of file EvFDaqDirector.h.

Referenced by unlockAndCloseMergeStream().

FILE* evf::EvFDaqDirector::data_rw_stream
private
bool evf::EvFDaqDirector::directorBu_
private

Definition at line 150 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

DirManager evf::EvFDaqDirector::dirManager_
private

Definition at line 181 of file EvFDaqDirector.h.

Referenced by findCurrentRunDir(), and preBeginRun().

bool evf::EvFDaqDirector::emptyLumisectionMode_
private

Definition at line 157 of file EvFDaqDirector.h.

Referenced by emptyLumisectionMode(), and EvFDaqDirector().

unsigned int evf::EvFDaqDirector::eolsNFilesIndex_ = 1
private

Definition at line 211 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

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

Definition at line 202 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

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

Definition at line 203 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

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

Definition at line 198 of file EvFDaqDirector.h.

Referenced by bumpFile(), and setFMS().

int evf::EvFDaqDirector::fu_readwritelock_fd_
private

Definition at line 169 of file EvFDaqDirector.h.

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

struct flock evf::EvFDaqDirector::fu_rw_flk
private

Definition at line 189 of file EvFDaqDirector.h.

Referenced by updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_fulk
private

Definition at line 190 of file EvFDaqDirector.h.

Referenced by updateFuLock().

FILE* evf::EvFDaqDirector::fu_rw_lock_stream
private
int evf::EvFDaqDirector::fulocal_rwlock_fd2_
private

Definition at line 172 of file EvFDaqDirector.h.

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

int evf::EvFDaqDirector::fulocal_rwlock_fd_
private

Definition at line 171 of file EvFDaqDirector.h.

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

unsigned int evf::EvFDaqDirector::fuLockPollInterval_
private

Definition at line 156 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateFuLock().

std::string evf::EvFDaqDirector::hltSourceDirectory_
private

Definition at line 155 of file EvFDaqDirector.h.

Referenced by initRun().

std::string evf::EvFDaqDirector::hostname_
private
pthread_mutex_t evf::EvFDaqDirector::init_lock_ = PTHREAD_MUTEX_INITIALIZER
private

Definition at line 205 of file EvFDaqDirector.h.

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

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

Definition at line 216 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet(), and getStreamMergeType().

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

Definition at line 219 of file EvFDaqDirector.h.

Referenced by getStreamMergeType().

std::string evf::EvFDaqDirector::mergeTypePset_
private

Definition at line 159 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet().

bool evf::EvFDaqDirector::microMergeDisabled_
private

Definition at line 158 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and microMergeDisabled().

unsigned int evf::EvFDaqDirector::nStreams_ =0
private

Definition at line 207 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

unsigned int evf::EvFDaqDirector::nThreads_ =0
private

Definition at line 208 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

bool evf::EvFDaqDirector::outputAdler32Recheck_
private

Definition at line 152 of file EvFDaqDirector.h.

Referenced by outputAdler32Recheck().

unsigned long evf::EvFDaqDirector::previousFileSize_
private

Definition at line 183 of file EvFDaqDirector.h.

Referenced by bumpFile().

bool evf::EvFDaqDirector::readEolsDefinition_ = true
private

Definition at line 210 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

bool evf::EvFDaqDirector::requireTSPSet_
private

Definition at line 153 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().

unsigned int evf::EvFDaqDirector::run_
private
std::string evf::EvFDaqDirector::run_dir_
private
std::string evf::EvFDaqDirector::run_string_
private

Definition at line 162 of file EvFDaqDirector.h.

Referenced by initRun().

std::string evf::EvFDaqDirector::selectedTransferMode_
private

Definition at line 154 of file EvFDaqDirector.h.

Referenced by getStreamDestinations().

unsigned int evf::EvFDaqDirector::stop_ls_override_ = 0
private

Definition at line 213 of file EvFDaqDirector.h.

Referenced by updateFuLock().

std::string evf::EvFDaqDirector::stopFilePath_
private

Definition at line 212 of file EvFDaqDirector.h.

Referenced by initRun(), and updateFuLock().

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

Definition at line 199 of file EvFDaqDirector.h.

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

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

Definition at line 215 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().