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

Detailed Description

Definition at line 51 of file EvFDaqDirector.h.

Member Enumeration Documentation

Constructor & Destructor Documentation

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

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

44  :
45  base_dir_(pset.getUntrackedParameter<std::string> ("baseDir", ".")),
46  bu_base_dir_(pset.getUntrackedParameter<std::string> ("buBaseDir", ".")),
47  directorBu_(pset.getUntrackedParameter<bool> ("directorIsBu", false)),
48  run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
49  outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck",false)),
50  requireTSPSet_(pset.getUntrackedParameter<bool>("requireTransfersPSet",false)),
51  selectedTransferMode_(pset.getUntrackedParameter<std::string>("selectedTransferMode","")),
52  hltSourceDirectory_(pset.getUntrackedParameter<std::string>("hltSourceDirectory","")),
53  fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>("fuLockPollInterval",2000)),
54  emptyLumisectionMode_(pset.getUntrackedParameter<bool>("emptyLumisectionMode",false)),
55  microMergeDisabled_(pset.getUntrackedParameter<bool>("microMergeDisabled",false)),
56  hostname_(""),
57  bu_readlock_fd_(-1),
58  bu_writelock_fd_(-1),
63 
67  //bu_w_monitor_stream(0),
68  //bu_t_monitor_stream(0),
69  data_rw_stream(0),
70 
72 
74 
75  bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
76  bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
77  bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
78  bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
79  fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
80  fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
81  data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
82  data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
83  //fulocal_rw_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
84  //fulocal_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
85  //fulocal_rw_flk2( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
86  //fulocal_rw_fulk2( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
87  {
88 
95 
96  //save hostname for later
97  char hostname[33];
98  gethostname(hostname,32);
99  hostname_ = hostname;
100 
101  char * fuLockPollIntervalPtr = getenv("FFF_LOCKPOLLINTERVAL");
102  if (fuLockPollIntervalPtr) {
103  try {
104  fuLockPollInterval_=boost::lexical_cast<unsigned int>(std::string(fuLockPollIntervalPtr));
105  edm::LogInfo("EvFDaqDirector") << "Setting fu lock poll interval by environment string: " << fuLockPollInterval_ << " us";
106  }
107  catch( boost::bad_lexical_cast const& ) {
108  edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fuLockPollIntervalPtr);
109  }
110  }
111 
112  char * emptyLumiModePtr = getenv("FFF_EMPTYLSMODE");
113  if (emptyLumiModePtr) {
114  emptyLumisectionMode_ = true;
115  edm::LogInfo("EvFDaqDirector") << "Setting empty lumisection mode";
116  }
117 
118  char * microMergeDisabledPtr = getenv("FFF_MICROMERGEDISABLED");
119  if (microMergeDisabledPtr) {
120  microMergeDisabled_ = true;
121  edm::LogInfo("EvFDaqDirector") << "Disabling dat file micro-merge by the HLT process (delegated to the hlt daemon)";
122  }
123  }
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 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 252 of file EvFDaqDirector.cc.

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

253  {
254  if (fulocal_rwlock_fd_!=-1) {
255  unlockFULocal();
256  close(fulocal_rwlock_fd_);
257  }
258 
259  if (fulocal_rwlock_fd2_!=-1) {
260  unlockFULocal2();
261  close(fulocal_rwlock_fd2_);
262  }
263 
264  }

Member Function Documentation

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

Definition at line 69 of file EvFDaqDirector.h.

References run_dir_.

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

Definition at line 70 of file EvFDaqDirector.h.

References bu_run_dir_.

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

Definition at line 71 of file EvFDaqDirector.h.

References bu_run_open_dir_.

71 {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 711 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().

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

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

898  {
899  if(transferSystemJson_) return;
900 
901  transferSystemJson_.reset(new Json::Value);
902  edm::ParameterSet const& topPset = edm::getParameterSet(pc.parameterSetID());
903  if (topPset.existsAs<edm::ParameterSet>("transferSystem",true))
904  {
905  const edm::ParameterSet& tsPset(topPset.getParameterSet("transferSystem"));
906 
907  Json::Value destinationsVal(Json::arrayValue);
908  std::vector<std::string> destinations = tsPset.getParameter<std::vector<std::string>>("destinations");
909  for (auto & dest: destinations) destinationsVal.append(dest);
910  (*transferSystemJson_)["destinations"]=destinationsVal;
911 
912  Json::Value modesVal(Json::arrayValue);
913  std::vector<std::string> modes = tsPset.getParameter< std::vector<std::string> >("transferModes");
914  for (auto & mode: modes) modesVal.append(mode);
915  (*transferSystemJson_)["transferModes"]=modesVal;
916 
917  for (auto psKeyItr =tsPset.psetTable().begin();psKeyItr!=tsPset.psetTable().end(); ++ psKeyItr) {
918  if (psKeyItr->first!="destinations" && psKeyItr->first!="transferModes") {
919  const edm::ParameterSet & streamDef = tsPset.getParameterSet(psKeyItr->first);
920  Json::Value streamVal;
921  for (auto & mode : modes) {
922  //validation
923  if (!streamDef.existsAs<std::vector<std::string>>(mode,true))
924  throw cms::Exception("EvFDaqDirector") << " Missing transfer system specification for -:" << psKeyItr->first << " (transferMode " << mode << ")";
925  std::vector<std::string> streamDestinations = streamDef.getParameter<std::vector<std::string>>(mode);
926 
927  Json::Value sDestsValue(Json::arrayValue);
928 
929  if (!streamDestinations.size())
930  throw cms::Exception("EvFDaqDirector") << " Missing transter system destination(s) for -: "<< psKeyItr->first << ", mode:" << mode;
931 
932  for (auto & sdest:streamDestinations) {
933  bool sDestValid=false;
934  sDestsValue.append(sdest);
935  for (auto & dest: destinations) {
936  if (dest==sdest) sDestValid=true;
937  }
938  if (!sDestValid)
939  throw cms::Exception("EvFDaqDirector") << " Invalid transter system destination specified for -: "<< psKeyItr->first << ", mode:" << mode << ", dest:"<<sdest;
940  }
941  streamVal[mode]=sDestsValue;
942  }
943  (*transferSystemJson_)[psKeyItr->first] = streamVal;
944  }
945  }
946  }
947  else {
948  if (requireTSPSet_)
949  throw cms::Exception("EvFDaqDirector") << "transferSystem PSet not found";
950  }
951  }
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 1000 of file EvFDaqDirector.cc.

References run_dir_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::checkNextEvent().

1000  {
1001  std::string proc_flag = run_dir_ + "/processing";
1002  int proc_flag_fd = open(proc_flag.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
1003  close(proc_flag_fd);
1004  }
void evf::EvFDaqDirector::createRunOpendirMaybe ( )

Definition at line 869 of file EvFDaqDirector.cc.

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

Referenced by initRun().

869  {
870  // create open dir if not already there
871 
873  if (!boost::filesystem::is_directory(openPath)) {
874  LogDebug("EvFDaqDirector") << "<open> FU dir not found. Creating... -:" << openPath.string();
875  boost::filesystem::create_directories(openPath);
876  }
877  }
#define LogDebug(id)
std::string getRunOpenDirPath() const
bool evf::EvFDaqDirector::emptyLumisectionMode ( ) const
inline

Definition at line 126 of file EvFDaqDirector.h.

References emptyLumisectionMode_.

Referenced by FedRawDataInputSource::checkNextEvent().

126 {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 278 of file EvFDaqDirector.cc.

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

279  {
281  desc.setComment("Service used for file locking arbitration and for propagating information between other EvF components");
282  desc.addUntracked<std::string> ("baseDir", ".")->setComment("Local base directory for run output");
283  desc.addUntracked<std::string> ("buBaseDir", ".")->setComment("BU base ramdisk directory ");
284  desc.addUntracked<unsigned int> ("runNumber",0)->setComment("Run Number in ramdisk to open");
285  desc.addUntracked<bool>("outputAdler32Recheck",false)->setComment("Check Adler32 of per-process output files while micro-merging");
286  desc.addUntracked<bool>("requireTransfersPSet",false)->setComment("Require complete transferSystem PSet in the process configuration");
287  desc.addUntracked<std::string>("selectedTransferMode","")->setComment("Selected transfer mode (choice in Lvl0 propagated as Python parameter");
288  desc.addUntracked<unsigned int>("fuLockPollInterval",2000)->setComment("Lock polling interval in microseconds for the input directory file lock");
289  desc.addUntracked<bool>("emptyLumisectionMode",false)->setComment("Enables writing stream output metadata even when no events are processed in a lumisection");
290  desc.addUntracked<bool>("microMergeDisabled",false)->setComment("Disabled micro-merging by the Output Module, so it is later done by hltd service");
291  desc.setAllowAnything();
292  descriptions.add("EvFDaqDirector", desc);
293  }
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 73 of file EvFDaqDirector.h.

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

73 { 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 451 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::createBoLSFile().

451  {
452  return run_dir_ + "/" + fffnaming::bolsFileName(run_,ls);
453  }
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 387 of file EvFDaqDirector.cc.

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

387  {
389  }
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 443 of file EvFDaqDirector.cc.

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

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

443  {
444  return bu_run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
445  }
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 447 of file EvFDaqDirector.cc.

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

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

447  {
448  return run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
449  }
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 455 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

455  {
456  return bu_run_dir_ + "/" + fffnaming::eorFileName(run_);
457  }
std::string eorFileName(const unsigned int run)
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getEoRFilePathOnFU ( ) const

Definition at line 460 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::checkNextEvent().

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

Definition at line 415 of file EvFDaqDirector.cc.

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

Referenced by DQMFileSaver::globalBeginRun().

415  {
416  return run_dir_ + "/" + fffnaming::initFileNameWithPid(run_,0,stream);
417  }
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 370 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

370  {
372  }
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 407 of file EvFDaqDirector.cc.

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

407  {
409  }
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 403 of file EvFDaqDirector.cc.

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

Referenced by maybeCreateAndLockFileHeadForStream().

403  {
405  }
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 427 of file EvFDaqDirector.cc.

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

427  {
429  }
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 439 of file EvFDaqDirector.cc.

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

439  {
441  }
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 650 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().

650  {
651 
652  boost::filesystem::ifstream ij(BUEoLSFile);
653  Json::Value deserializeRoot;
655 
656  if (!reader.parse(ij, deserializeRoot)) {
657  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << BUEoLSFile;
658  return -1;
659  }
660 
662  DataPoint dp;
663  dp.deserialize(deserializeRoot);
664 
665  //read definition
666  if (readEolsDefinition_) {
667  //std::string def = boost::algorithm::trim(dp.getDefinition());
669  if (!def.size()) readEolsDefinition_=false;
670  while (def.size()) {
671  std::string fullpath;
672  if (def.find('/')==0)
673  fullpath = def;
674  else
675  fullpath = bu_run_dir_+'/'+def;
676  struct stat buf;
677  if (stat(fullpath.c_str(), &buf) == 0) {
678  DataPointDefinition eolsDpd;
679  std::string defLabel = "legend";
680  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
681  if (eolsDpd.getNames().size()==0) {
682  //try with "data" label if "legend" format is not used
683  eolsDpd = DataPointDefinition();
684  defLabel="data";
685  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
686  }
687  for (unsigned int i=0;i<eolsDpd.getNames().size();i++)
688  if (eolsDpd.getNames().at(i)=="NFiles")
690  readEolsDefinition_=false;
691  break;
692  }
693  //check if we can still find definition
694  if (def.size()<=1 || def.find('/')==std::string::npos) {
695  readEolsDefinition_=false;
696  break;
697  }
698  def = def.substr(def.find('/')+1);
699  }
700  }
701 
702  if (dp.getData().size()>eolsNFilesIndex_)
703  data = dp.getData()[eolsNFilesIndex_];
704  else {
705  edm::LogError("EvFDaqDirector") << " error reading number of files from BU JSON -: " << BUEoLSFile;
706  return -1;
707  }
708  return boost::lexical_cast<int>(data);
709  }
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 391 of file EvFDaqDirector.cc.

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

391  {
392  return run_dir_ + "/open/" + fffnaming::streamerDataFileNameWithPid(run_,ls,stream);
393  }
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 411 of file EvFDaqDirector.cc.

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

411  {
412  return run_dir_ + "/open/" + fffnaming::initFileNameWithPid(run_,0,stream);
413  }
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 383 of file EvFDaqDirector.cc.

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

383  {
385  }
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 395 of file EvFDaqDirector.cc.

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

395  {
396  return run_dir_ + "/open/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
397  }
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 419 of file EvFDaqDirector.cc.

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

419  {
421  }
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 379 of file EvFDaqDirector.cc.

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

379  {
380  return bu_run_dir_ + "/open/" + fffnaming::inputRawFileName(run_,ls,index);
381  }
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 431 of file EvFDaqDirector.cc.

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

431  {
432  return run_dir_ + "/open/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
433  }
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 399 of file EvFDaqDirector.cc.

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

399  {
401  }
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 423 of file EvFDaqDirector.cc.

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

423  {
425  }
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 375 of file EvFDaqDirector.cc.

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

Referenced by removeFile().

375  {
377  }
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 435 of file EvFDaqDirector.cc.

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

435  {
437  }
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 104 of file EvFDaqDirector.h.

References run_.

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

Definition at line 97 of file EvFDaqDirector.h.

References run_dir_.

Referenced by createRunOpendirMaybe(), and initRun().

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

Definition at line 953 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_.

954  {
955  std::string streamRequestName;
956  if (transferSystemJson_->isMember(stream.c_str()))
957  streamRequestName = stream;
958  else {
959  std::stringstream msg;
960  msg << "Transfer system mode definitions missing for -: " << stream;
961  if (requireTSPSet_)
962  throw cms::Exception("EvFDaqDirector") << msg.str();
963  else {
964  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
965  return std::string("Failsafe");
966  }
967  }
968  //return empty if strict check parameter is not on
970  edm::LogWarning("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter."
971  << "Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
972  return std::string("Failsafe");
973  }
975  throw cms::Exception("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter.";
976  }
977  //check if stream has properly listed transfer stream
978  if (!transferSystemJson_->get(streamRequestName, "").isMember(selectedTransferMode_.c_str()))
979  {
980  std::stringstream msg;
981  msg << "Selected transfer mode " << selectedTransferMode_ << " is not specified for stream " << streamRequestName;
982  if (requireTSPSet_)
983  throw cms::Exception("EvFDaqDirector") << msg.str();
984  else
985  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
986  return std::string("Failsafe");
987  }
988  Json::Value destsVec = transferSystemJson_->get(streamRequestName, "").get(selectedTransferMode_,"");
989 
990  //flatten string json::Array into CSV std::string
992  for (Json::Value::iterator it = destsVec.begin(); it!=destsVec.end(); it++)
993  {
994  if (ret!="") ret +=",";
995  ret+=(*it).asString();
996  }
997  return ret;
998  }
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 111 of file EvFDaqDirector.h.

References streamFileTracker_.

Referenced by FedRawDataInputSource::getNextEvent().

111 {return &streamFileTracker_;}
std::vector< int > streamFileTracker_
std::string evf::EvFDaqDirector::initFileName ( std::string const &  stream) const
private
void evf::EvFDaqDirector::initRun ( void  )

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

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

References nStreams_, and nThreads_.

Referenced by FedRawDataInputSource::getNextEvent().

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

Definition at line 847 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by updateFuLock().

847  {
848  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_flk);
849  flock(fulocal_rwlock_fd_,LOCK_SH);
850  }
void evf::EvFDaqDirector::lockFULocal2 ( )

Definition at line 858 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

858  {
859  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_flk2);
860  flock(fulocal_rwlock_fd2_,LOCK_EX);
861  }
void evf::EvFDaqDirector::lockInitLock ( )

Definition at line 839 of file EvFDaqDirector.cc.

References init_lock_.

839  {
840  pthread_mutex_lock(&init_lock_);
841  }
pthread_mutex_t init_lock_
FILE * evf::EvFDaqDirector::maybeCreateAndLockFileHeadForStream ( unsigned int  ls,
std::string &  stream 
)

Definition at line 819 of file EvFDaqDirector.cc.

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

819  {
820  data_rw_stream = fopen(getMergedDatFilePath(ls,stream).c_str(), "a"); //open stream for appending
822  if (data_readwrite_fd_ == -1)
823  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for datamerge "
824  << strerror(errno);
825  else
826  LogDebug("EvFDaqDirector") << "creating filedesc for datamerge -: "
828  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_flk);
829 
830  return data_rw_stream;
831  }
#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 127 of file EvFDaqDirector.h.

References microMergeDisabled_.

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

Definition at line 800 of file EvFDaqDirector.cc.

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

Referenced by initRun().

800  {
801  if (create) {
802  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
803  S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
804  chmod(fulockfile.c_str(),0766);
805  } else {
806  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
807  }
808  if (fu_readwritelock_fd_ == -1)
809  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
810  << " create:" << create << " error:" << strerror(errno);
811  else
812  LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: "
814 
815  fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
816  }
#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 98 of file EvFDaqDirector.h.

References 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 68 of file EvFDaqDirector.h.

References DTTTrigCorrFirst::run, and run_.

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

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

312  {
313  close(bu_readlock_fd_);
314  close(bu_writelock_fd_);
315  if (directorBu_) {
316  std::string filename = bu_run_dir_ + "/bu.lock";
317  removeFile(filename);
318  }
319  }
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 267 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

267  {
268 
269  initRun();
270 
271  for (unsigned int i=0;i<bounds.maxNumberOfStreams();i++){
272  streamFileTracker_.push_back(-1);
273  }
274  nThreads_=bounds.maxNumberOfStreams();
275  nStreams_=bounds.maxNumberOfThreads();
276  }
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 295 of file EvFDaqDirector.cc.

References checkTransferSystemPSet().

Referenced by EvFDaqDirector().

296  {
298  }
void checkTransferSystemPSet(edm::ProcessContext const &pc)
void evf::EvFDaqDirector::preBeginRun ( edm::GlobalContext const &  globalContext)

Definition at line 300 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

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

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

322  {
323  //delete all files belonging to just closed lumi
324  unsigned int ls = globalContext.luminosityBlockID().luminosityBlock();
326  edm::LogWarning("EvFDaqDirector") << " Handles to check for files to delete were not set by the input source...";
327  return;
328  }
329 
330  std::unique_lock<std::mutex> lkw(*fileDeleteLockPtr_);
331  auto it = filesToDeletePtr_->begin();
332  while (it!=filesToDeletePtr_->end()) {
333  if (it->second->lumi_ == ls) {
334  const boost::filesystem::path filePath(it->second->fileName_);
335  LogDebug("EvFDaqDirector") << "Deleting input file -:" << it->second->fileName_;
336  try {
337  //rarely this fails but file gets deleted
338  boost::filesystem::remove(filePath);
339  }
340  catch (const boost::filesystem::filesystem_error& ex)
341  {
342  edm::LogError("EvFDaqDirector") << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() << ". Trying again.";
343  usleep(10000);
344  try {
345  boost::filesystem::remove(filePath);
346  }
347  catch (const boost::filesystem::filesystem_error&) {/*file gets deleted first time but exception is still thrown*/}
348  }
349  catch (std::exception& ex)
350  {
351  edm::LogError("EvFDaqDirector") << " - deleteFile std::exception CAUGHT -: " << ex.what() << ". Trying again.";
352  usleep(10000);
353  try {
354  boost::filesystem::remove(filePath);
355  } catch (std::exception&) {/*file gets deleted first time but exception is still thrown*/}
356  }
357 
358  delete it->second;
359  it = filesToDeletePtr_->erase(it);
360  }
361  else it++;
362  }
363  }
#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 365 of file EvFDaqDirector.cc.

References currentFileIndex_, and streamFileTracker_.

Referenced by EvFDaqDirector().

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

Definition at line 880 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

880  {
881 
882  boost::filesystem::ifstream ij(file);
883  Json::Value deserializeRoot;
885 
886  if (!reader.parse(ij, deserializeRoot)) {
887  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << file;
888  return -1;
889  }
890 
891  int ret = deserializeRoot.get("lastLS","").asInt();
892  return ret;
893 
894  }
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 471 of file EvFDaqDirector.cc.

References getRawFilePath().

Referenced by postEndRun().

471  {
473  }
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 464 of file EvFDaqDirector.cc.

464  {
465  int retval = remove(filename.c_str());
466  if (retval != 0)
467  edm::LogError("EvFDaqDirector") << "Could not remove used file -: " << filename << ". error = "
468  << strerror(errno);
469  }
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 120 of file EvFDaqDirector.h.

References fileDeleteLockPtr_, and filesToDeletePtr_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

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

Definition at line 109 of file EvFDaqDirector.h.

References fms_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

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

Definition at line 789 of file EvFDaqDirector.cc.

References fu_rw_lock_stream, and getHLTprescales::readIndex().

Referenced by initRun().

789  {
790  if (fu_rw_lock_stream == 0)
791  edm::LogError("EvFDaqDirector") << "Error creating fu read/write lock stream "
792  << strerror(errno);
793  else {
794  edm::LogInfo("EvFDaqDirector") << "Initializing FU LOCK FILE";
795  unsigned int readLs = 1, readIndex = 0;
796  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
797  }
798  }
void evf::EvFDaqDirector::unlockAndCloseMergeStream ( )

Definition at line 833 of file EvFDaqDirector.cc.

References data_readwrite_fd_, data_rw_fulk, and data_rw_stream.

833  {
834  fflush(data_rw_stream);
835  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_fulk);
836  fclose(data_rw_stream);
837  }
struct flock data_rw_fulk
void evf::EvFDaqDirector::unlockFULocal ( )

Definition at line 852 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

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

852  {
853  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_fulk);
854  flock(fulocal_rwlock_fd_,LOCK_UN);
855  }
void evf::EvFDaqDirector::unlockFULocal2 ( )

Definition at line 863 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

863  {
864  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_fulk2);
865  flock(fulocal_rwlock_fd2_,LOCK_UN);
866  }
void evf::EvFDaqDirector::unlockInitLock ( )

Definition at line 843 of file EvFDaqDirector.cc.

References init_lock_.

843  {
844  pthread_mutex_unlock(&init_lock_);
845  }
pthread_mutex_t init_lock_
void evf::EvFDaqDirector::updateFileIndex ( int const &  fileIndex)
inline

Definition at line 110 of file EvFDaqDirector.h.

References currentFileIndex_.

Referenced by FedRawDataInputSource::getNextEvent().

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

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

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

Referenced by initRun().

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

Definition at line 144 of file EvFDaqDirector.h.

Referenced by initRun().

struct flock evf::EvFDaqDirector::bu_r_flk
private

Definition at line 180 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_r_fulk
private

Definition at line 182 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_r_lock_stream
private

Definition at line 169 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_readlock_fd_
private

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

Referenced by buBaseRunOpenDir(), and initRun().

FILE* evf::EvFDaqDirector::bu_t_monitor_stream
private

Definition at line 172 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_flk
private

Definition at line 179 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_fulk
private

Definition at line 181 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_w_lock_stream
private

Definition at line 168 of file EvFDaqDirector.h.

Referenced by initRun().

FILE* evf::EvFDaqDirector::bu_w_monitor_stream
private

Definition at line 171 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_writelock_fd_
private

Definition at line 162 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

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

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

Referenced by maybeCreateAndLockFileHeadForStream().

struct flock evf::EvFDaqDirector::data_rw_fulk
private

Definition at line 186 of file EvFDaqDirector.h.

Referenced by unlockAndCloseMergeStream().

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

Definition at line 145 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

DirManager evf::EvFDaqDirector::dirManager_
private

Definition at line 175 of file EvFDaqDirector.h.

Referenced by findCurrentRunDir(), and preBeginRun().

bool evf::EvFDaqDirector::emptyLumisectionMode_
private

Definition at line 152 of file EvFDaqDirector.h.

Referenced by emptyLumisectionMode(), and EvFDaqDirector().

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

Definition at line 205 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

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

Definition at line 196 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

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

Definition at line 197 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

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

Definition at line 192 of file EvFDaqDirector.h.

Referenced by bumpFile(), and setFMS().

int evf::EvFDaqDirector::fu_readwritelock_fd_
private

Definition at line 163 of file EvFDaqDirector.h.

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

struct flock evf::EvFDaqDirector::fu_rw_flk
private

Definition at line 183 of file EvFDaqDirector.h.

Referenced by updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_fulk
private

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

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

int evf::EvFDaqDirector::fulocal_rwlock_fd_
private

Definition at line 165 of file EvFDaqDirector.h.

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

unsigned int evf::EvFDaqDirector::fuLockPollInterval_
private

Definition at line 151 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateFuLock().

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

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

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

bool evf::EvFDaqDirector::microMergeDisabled_
private

Definition at line 153 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and microMergeDisabled().

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

Definition at line 201 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

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

Definition at line 202 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

bool evf::EvFDaqDirector::outputAdler32Recheck_
private

Definition at line 147 of file EvFDaqDirector.h.

Referenced by outputAdler32Recheck().

unsigned long evf::EvFDaqDirector::previousFileSize_
private

Definition at line 177 of file EvFDaqDirector.h.

Referenced by bumpFile().

bool evf::EvFDaqDirector::readEolsDefinition_ = true
private

Definition at line 204 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

bool evf::EvFDaqDirector::requireTSPSet_
private

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

Referenced by initRun().

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

Definition at line 149 of file EvFDaqDirector.h.

Referenced by getStreamDestinations().

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

Definition at line 207 of file EvFDaqDirector.h.

Referenced by updateFuLock().

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

Definition at line 206 of file EvFDaqDirector.h.

Referenced by initRun(), and updateFuLock().

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

Definition at line 193 of file EvFDaqDirector.h.

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

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

Definition at line 209 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().