CMS 3D CMS Logo

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::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)
 
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 ()
 
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 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::string stopFilePathPid_
 
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(), AlCaHLTBitMon_QueryRunRegistry::string, edm::ActivityRegistry::watchPostGlobalEndRun(), edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreBeginJob(), edm::ActivityRegistry::watchPreGlobalBeginRun(), and edm::ActivityRegistry::watchPreGlobalEndLumi().

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 
99 
100  //save hostname for later
101  char hostname[33];
102  gethostname(hostname,32);
103  hostname_ = hostname;
104 
105  char * fuLockPollIntervalPtr = getenv("FFF_LOCKPOLLINTERVAL");
106  if (fuLockPollIntervalPtr) {
107  try {
108  fuLockPollInterval_=boost::lexical_cast<unsigned int>(std::string(fuLockPollIntervalPtr));
109  edm::LogInfo("EvFDaqDirector") << "Setting fu lock poll interval by environment string: " << fuLockPollInterval_ << " us";
110  }
111  catch( boost::bad_lexical_cast const& ) {
112  edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fuLockPollIntervalPtr);
113  }
114  }
115 
116  char * emptyLumiModePtr = getenv("FFF_EMPTYLSMODE");
117  if (emptyLumiModePtr) {
118  emptyLumisectionMode_ = true;
119  edm::LogInfo("EvFDaqDirector") << "Setting empty lumisection mode";
120  }
121 
122  char * microMergeDisabledPtr = getenv("FFF_MICROMERGEDISABLED");
123  if (microMergeDisabledPtr) {
124  microMergeDisabled_ = true;
125  edm::LogInfo("EvFDaqDirector") << "Disabling dat file micro-merge by the HLT process (delegated to the hlt daemon)";
126  }
127  }
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 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 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 259 of file EvFDaqDirector.cc.

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

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

Member Function Documentation

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

Definition at line 71 of file EvFDaqDirector.h.

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

Definition at line 72 of file EvFDaqDirector.h.

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

Definition at line 73 of file EvFDaqDirector.h.

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

References evf::FastMonitoringService::accumulateFileSize(), fms_, getEoLSFilePathOnBU(), getInputJsonFilePath(), getNFilesFromEoLS(), diffTreeTool::index, eostools::ls(), previousFileSize_, trackingPlots::stat, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by updateFuLock().

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

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

Referenced by preBeginJob().

1006  {
1007  if (mergeTypePset_.empty()) return;
1008  if(mergeTypeMap_.size()) return;
1009  edm::ParameterSet const& topPset = edm::getParameterSet(pc.parameterSetID());
1010  if (topPset.existsAs<edm::ParameterSet>(mergeTypePset_,true))
1011  {
1012  const edm::ParameterSet& tsPset(topPset.getParameterSet(mergeTypePset_));
1013  for (std::string pname : tsPset.getParameterNames()) {
1014  std::string streamType = tsPset.getParameter<std::string>(pname);
1015  mergeTypeMap_[pname]=streamType;
1016  }
1017  }
1018  }
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 902 of file EvFDaqDirector.cc.

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

Referenced by preBeginJob().

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

References run_dir_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource::checkNextEvent().

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

Definition at line 874 of file EvFDaqDirector.cc.

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

Referenced by initRun().

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

Definition at line 128 of file EvFDaqDirector.h.

Referenced by FedRawDataInputSource::checkNextEvent().

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 282 of file EvFDaqDirector.cc.

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

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

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

75 { 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(path, rec=False)
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(path, rec=False)
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(path, rec=False)
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(path, rec=False)
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 371 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

371  {
373  }
std::string inputJsonFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
def ls(path, rec=False)
Definition: eostools.py:348
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(path, rec=False)
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  }
std::string streamerDataFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
def ls(path, rec=False)
Definition: eostools.py:348
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(path, rec=False)
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(path, rec=False)
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 655 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

655  {
656 
657  boost::filesystem::ifstream ij(BUEoLSFile);
658  Json::Value deserializeRoot;
660 
661  if (!reader.parse(ij, deserializeRoot)) {
662  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << BUEoLSFile;
663  return -1;
664  }
665 
667  DataPoint dp;
668  dp.deserialize(deserializeRoot);
669 
670  //read definition
671  if (readEolsDefinition_) {
672  //std::string def = boost::algorithm::trim(dp.getDefinition());
674  if (!def.size()) readEolsDefinition_=false;
675  while (def.size()) {
676  std::string fullpath;
677  if (def.find('/')==0)
678  fullpath = def;
679  else
680  fullpath = bu_run_dir_+'/'+def;
681  struct stat buf;
682  if (stat(fullpath.c_str(), &buf) == 0) {
683  DataPointDefinition eolsDpd;
684  std::string defLabel = "legend";
685  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
686  if (eolsDpd.getNames().size()==0) {
687  //try with "data" label if "legend" format is not used
688  eolsDpd = DataPointDefinition();
689  defLabel="data";
690  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
691  }
692  for (unsigned int i=0;i<eolsDpd.getNames().size();i++)
693  if (eolsDpd.getNames().at(i)=="NFiles")
695  readEolsDefinition_=false;
696  break;
697  }
698  //check if we can still find definition
699  if (def.size()<=1 || def.find('/')==std::string::npos) {
700  readEolsDefinition_=false;
701  break;
702  }
703  def = def.substr(def.find('/')+1);
704  }
705  }
706 
707  if (dp.getData().size()>eolsNFilesIndex_)
708  data = dp.getData()[eolsNFilesIndex_];
709  else {
710  edm::LogError("EvFDaqDirector") << " error reading number of files from BU JSON -: " << BUEoLSFile;
711  return -1;
712  }
713  return boost::lexical_cast<int>(data);
714  }
std::vector< std::string > & getData()
Definition: DataPoint.h:58
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_
char data[epos_bytes_allocation]
Definition: EPOS_Wrapper.h:82
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(path, rec=False)
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  }
std::string inputJsonFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
def ls(path, rec=False)
Definition: eostools.py:348
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  }
std::string streamerJsonFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
def ls(path, rec=False)
Definition: eostools.py:348
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(path, rec=False)
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(path, rec=False)
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(path, rec=False)
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  }
std::string streamerJsonFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
def ls(path, rec=False)
Definition: eostools.py:348
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(path, rec=False)
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(path, rec=False)
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(path, rec=False)
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 106 of file EvFDaqDirector.h.

References AlCaHLTBitMon_QueryRunRegistry::string.

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

Definition at line 99 of file EvFDaqDirector.h.

Referenced by createRunOpendirMaybe(), and initRun().

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

Definition at line 958 of file EvFDaqDirector.cc.

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

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

Definition at line 1020 of file EvFDaqDirector.cc.

References mergeTypeMap_, MergeTypeNames_, and AlCaHLTBitMon_QueryRunRegistry::string.

1021  {
1022  auto mergeTypeItr = mergeTypeMap_.find(stream.c_str());
1023  if (mergeTypeItr == mergeTypeMap_.end()) {
1024  edm::LogInfo("EvFDaqDirector") << " No merging type specified for stream " << stream << ". Using default value";
1025  assert(defaultType<MergeTypeNames_.size());
1026  std::string defaultName = MergeTypeNames_[defaultType];
1027  mergeTypeMap_[stream] = defaultName;
1028  return defaultName;
1029  }
1030  return mergeTypeItr->second;
1031  }
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 129 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_, trackingPlots::stat, stopFilePath_, stopFilePathPid_, AlCaHLTBitMon_QueryRunRegistry::string, and tryInitializeFuLockFile().

Referenced by preallocate().

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

Definition at line 852 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by updateFuLock().

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

Definition at line 863 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

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

Definition at line 844 of file EvFDaqDirector.cc.

References init_lock_.

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

Definition at line 824 of file EvFDaqDirector.cc.

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

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

Definition at line 805 of file EvFDaqDirector.cc.

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

Referenced by initRun().

805  {
806  if (create) {
807  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
808  S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
809  chmod(fulockfile.c_str(),0766);
810  } else {
811  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
812  }
813  if (fu_readwritelock_fd_ == -1)
814  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
815  << " create:" << create << " error:" << strerror(errno);
816  else
817  LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: "
819 
820  fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
821  }
#define LogDebug(id)
def create(alignables, pedeDump, additionalData, outputFile, config)
def chmod(path, mode)
Definition: eostools.py:293
bool evf::EvFDaqDirector::outputAdler32Recheck ( ) const
inline

Definition at line 100 of file EvFDaqDirector.h.

References AlCaHLTBitMon_QueryRunRegistry::string.

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

References findQualityFiles::run.

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

Definition at line 318 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

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

Definition at line 274 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

274  {
275 
276  initRun();
277 
278  nThreads_=bounds.maxNumberOfStreams();
279  nStreams_=bounds.maxNumberOfThreads();
280  }
unsigned int nThreads_
unsigned int nStreams_
void evf::EvFDaqDirector::preBeginJob ( edm::PathsAndConsumesOfModulesBase const &  ,
edm::ProcessContext const &  pc 
)

Definition at line 300 of file EvFDaqDirector.cc.

References checkMergeTypePSet(), and checkTransferSystemPSet().

Referenced by EvFDaqDirector().

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

Definition at line 306 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

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

Definition at line 327 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

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

Definition at line 885 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

885  {
886 
887  boost::filesystem::ifstream ij(file);
888  Json::Value deserializeRoot;
890 
891  if (!reader.parse(ij, deserializeRoot)) {
892  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << file;
893  return -1;
894  }
895 
896  int ret = deserializeRoot.get("lastLS","").asInt();
897  return ret;
898 
899  }
Value get(UInt index, const Value &defaultValue) const
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  }
std::string getRawFilePath(const unsigned int ls, const unsigned int index) const
void removeFile(unsigned int ls, unsigned int index)
def ls(path, rec=False)
Definition: eostools.py:348
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  }
void evf::EvFDaqDirector::setDeleteTracking ( std::mutex fileDeleteLock,
std::list< std::pair< int, InputFile * >> *  filesToDelete 
)
inline

Definition at line 120 of file EvFDaqDirector.h.

References AlCaHLTBitMon_QueryRunRegistry::string.

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

Referenced by FedRawDataInputSource::FedRawDataInputSource().

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

Definition at line 794 of file EvFDaqDirector.cc.

References fu_rw_lock_stream, and getHLTprescales::readIndex().

Referenced by initRun().

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

Definition at line 838 of file EvFDaqDirector.cc.

References data_readwrite_fd_, data_rw_fulk, and data_rw_stream.

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

Definition at line 857 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

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

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

Definition at line 868 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

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

Definition at line 848 of file EvFDaqDirector.cc.

References init_lock_.

848  {
849  pthread_mutex_unlock(&init_lock_);
850  }
pthread_mutex_t init_lock_
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(), trackerTree::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, trackingPlots::stat, stop_ls_override_, stopFilePath_, and stopFilePathPid_.

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

Member Data Documentation

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

Definition at line 145 of file EvFDaqDirector.h.

Referenced by initRun().

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

Definition at line 146 of file EvFDaqDirector.h.

Referenced by initRun().

struct flock evf::EvFDaqDirector::bu_r_flk
private

Definition at line 183 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_r_fulk
private

Definition at line 185 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_r_lock_stream
private

Definition at line 172 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_readlock_fd_
private

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

Referenced by initRun().

FILE* evf::EvFDaqDirector::bu_t_monitor_stream
private

Definition at line 175 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_flk
private

Definition at line 182 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_fulk
private

Definition at line 184 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_w_lock_stream
private

Definition at line 171 of file EvFDaqDirector.h.

Referenced by initRun().

FILE* evf::EvFDaqDirector::bu_w_monitor_stream
private

Definition at line 174 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_writelock_fd_
private

Definition at line 165 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

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

Definition at line 188 of file EvFDaqDirector.h.

Referenced by maybeCreateAndLockFileHeadForStream().

struct flock evf::EvFDaqDirector::data_rw_fulk
private

Definition at line 189 of file EvFDaqDirector.h.

Referenced by unlockAndCloseMergeStream().

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

Definition at line 147 of file EvFDaqDirector.h.

Referenced by initRun(), and postEndRun().

DirManager evf::EvFDaqDirector::dirManager_
private

Definition at line 178 of file EvFDaqDirector.h.

Referenced by preBeginRun().

bool evf::EvFDaqDirector::emptyLumisectionMode_
private

Definition at line 154 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

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

Definition at line 206 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

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

Definition at line 197 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi().

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

Definition at line 198 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi().

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

Definition at line 195 of file EvFDaqDirector.h.

Referenced by bumpFile().

int evf::EvFDaqDirector::fu_readwritelock_fd_
private

Definition at line 166 of file EvFDaqDirector.h.

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

struct flock evf::EvFDaqDirector::fu_rw_flk
private

Definition at line 186 of file EvFDaqDirector.h.

Referenced by updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_fulk
private

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

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

int evf::EvFDaqDirector::fulocal_rwlock_fd_
private

Definition at line 168 of file EvFDaqDirector.h.

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

unsigned int evf::EvFDaqDirector::fuLockPollInterval_
private

Definition at line 153 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and updateFuLock().

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

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

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

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

Definition at line 212 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet(), and getStreamMergeType().

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

Definition at line 215 of file EvFDaqDirector.h.

Referenced by getStreamMergeType().

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

Definition at line 156 of file EvFDaqDirector.h.

Referenced by checkMergeTypePSet().

bool evf::EvFDaqDirector::microMergeDisabled_
private

Definition at line 155 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

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

Definition at line 202 of file EvFDaqDirector.h.

Referenced by preallocate().

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

Definition at line 203 of file EvFDaqDirector.h.

Referenced by preallocate().

bool evf::EvFDaqDirector::outputAdler32Recheck_
private

Definition at line 149 of file EvFDaqDirector.h.

unsigned long evf::EvFDaqDirector::previousFileSize_
private

Definition at line 180 of file EvFDaqDirector.h.

Referenced by bumpFile().

bool evf::EvFDaqDirector::readEolsDefinition_ = true
private

Definition at line 205 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

bool evf::EvFDaqDirector::requireTSPSet_
private

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

Referenced by initRun().

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

Definition at line 151 of file EvFDaqDirector.h.

Referenced by getStreamDestinations().

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

Definition at line 209 of file EvFDaqDirector.h.

Referenced by updateFuLock().

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

Definition at line 207 of file EvFDaqDirector.h.

Referenced by initRun(), and updateFuLock().

std::string evf::EvFDaqDirector::stopFilePathPid_
private

Definition at line 208 of file EvFDaqDirector.h.

Referenced by initRun(), and updateFuLock().

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

Definition at line 211 of file EvFDaqDirector.h.

Referenced by checkTransferSystemPSet(), and getStreamDestinations().