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 | Private Member Functions | Private Attributes
evf::EvFDaqDirector Class Reference

#include <EvFDaqDirector.h>

Public Types

enum  FileStatus {
  noFile, sameFile, newFile, newLumi,
  runEnded
}
 

Public Member Functions

std::string & baseRunDir ()
 
std::string & buBaseRunDir ()
 
std::string & buBaseRunOpenDir ()
 
void createRunOpendirMaybe ()
 
 EvFDaqDirector (const edm::ParameterSet &pset, edm::ActivityRegistry &reg)
 
std::string findCurrentRunDir ()
 
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 getJumpFilePath () const
 
unsigned int getJumpIndex () const
 
unsigned int getJumpLS () 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 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::vector< int > * getStreamFileTracker ()
 
bool getTestModeNoBuilderUnit ()
 
bool isSingleStreamThread ()
 
void lockFULocal ()
 
void lockFULocal2 ()
 
void lockInitLock ()
 
FILE * maybeCreateAndLockFileHeadForStream (unsigned int ls, std::string &stream)
 
bool outputAdler32Recheck () const
 
void postEndRun (edm::GlobalContext const &globalContext)
 
void preallocate (edm::service::SystemBounds const &bounds)
 
void preBeginRun (edm::GlobalContext const &globalContext)
 
void preGlobalEndLumi (edm::GlobalContext const &globalContext)
 
void preSourceEvent (edm::StreamID const &streamID)
 
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)
 
 ~EvFDaqDirector ()
 

Private Member Functions

bool bumpFile (unsigned int &ls, unsigned int &index, std::string &nextFile, uint32_t &fsize)
 
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_
 
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_
 
std::string hostname_
 
pthread_mutex_t init_lock_ = PTHREAD_MUTEX_INITIALIZER
 
unsigned int jumpIndex_
 
unsigned int jumpLS_
 
unsigned int nStreams_ =0
 
unsigned int nThreads_ =0
 
bool outputAdler32Recheck_
 
unsigned long previousFileSize_
 
bool readEolsDefinition_ = true
 
unsigned int run_
 
std::string run_dir_
 
std::string run_string_
 
std::vector< int > streamFileTracker_
 
bool testModeNoBuilderUnit_
 

Detailed Description

Definition at line 38 of file EvFDaqDirector.h.

Member Enumeration Documentation

Constructor & Destructor Documentation

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

Definition at line 38 of file EvFDaqDirector.cc.

References base_dir_, bu_base_dir_, bu_run_dir_, bu_run_open_dir_, bu_w_lock_stream, bu_writelock_fd_, createRunOpendirMaybe(), directorBu_, edm::hlt::Exception, fu_readwritelock_fd_, fu_rw_lock_stream, fulocal_rwlock_fd2_, fulocal_rwlock_fd_, getRunOpenDirPath(), hostname_, init_lock_, NULL, openFULockfileStream(), postEndRun(), preallocate(), preBeginRun(), preGlobalEndLumi(), preSourceEvent(), run_, run_dir_, run_string_, contentValuesCheck::ss, AlCaHLTBitMon_QueryRunRegistry::string, tryInitializeFuLockFile(), edm::ActivityRegistry::watchPostGlobalEndRun(), edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreGlobalBeginRun(), edm::ActivityRegistry::watchPreGlobalEndLumi(), and edm::ActivityRegistry::watchPreSourceEvent().

39  :
41  pset.getUntrackedParameter<bool> ("testModeNoBuilderUnit",
42  false)
43  ),
44  base_dir_(
45  pset.getUntrackedParameter<std::string> ("baseDir", "/data")
46  ),
48  pset.getUntrackedParameter<std::string> ("buBaseDir", "/data")
49  ),
51  pset.getUntrackedParameter<bool> ("directorIsBu", false)
52  ),
53  run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
54  outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck",false)),
55  hostname_(""),
56  bu_readlock_fd_(-1),
57  bu_writelock_fd_(-1),
62 
66  //bu_w_monitor_stream(0),
67  //bu_t_monitor_stream(0),
68  data_rw_stream(0),
69 
71 
73  jumpLS_(0),
74  jumpIndex_(0),
75 
76  bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
77  bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
78  bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
79  bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
80  fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
81  fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
82  data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
83  data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
84  //fulocal_rw_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
85  //fulocal_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
86  //fulocal_rw_flk2( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
87  //fulocal_rw_fulk2( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
88  {
89 
95 
96  std::stringstream ss;
97  ss << "run" << std::setfill('0') << std::setw(6) << run_;
98  run_string_ = ss.str();
100 
101  //save hostname for later
102  char hostname[33];
103  gethostname(hostname,32);
104  hostname_ = hostname;
105  // check if base dir exists or create it accordingly
106  int retval = mkdir(base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
107  if (retval != 0 && errno != EEXIST) {
108  throw cms::Exception("DaqDirector") << " Error checking for base dir -: "
109  << base_dir_ << " mkdir error:" << strerror(errno);
110  }
111 
112  //create run dir in base dir
113  umask(0);
114  retval = mkdir(run_dir_.c_str(),
115  S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
116  if (retval != 0 && errno != EEXIST) {
117  throw cms::Exception("DaqDirector") << " Error creating run dir -: "
118  << run_dir_ << " mkdir error:" << strerror(errno);
119  }
120 
121  //create fu-local.lock in run open dir
122  if (!directorBu_) {
123 
125  std::string fulocal_lock_ = getRunOpenDirPath() +"/fu-local.lock";
126  fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
127  if (fulocal_rwlock_fd_==-1)
128  throw cms::Exception("DaqDirector") << " Error creating/opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
129  chmod(fulocal_lock_.c_str(),0777);
130  fsync(fulocal_rwlock_fd_);
131  //open second fd for another input source thread
132  fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
133  if (fulocal_rwlock_fd2_==-1)
134  throw cms::Exception("DaqDirector") << " Error opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
135  }
136 
137  //bu_run_dir: for FU, for which the base dir is local and the BU is remote, it is expected to be there
138  //for BU, it is created at this point
139  if (directorBu_)
140  {
142  std::string bulockfile = bu_run_dir_ + "/bu.lock";
143  std::string fulockfile = bu_run_dir_ + "/fu.lock";
144 
145  //make or find bu run dir
146  retval = mkdir(bu_run_dir_.c_str(),
147  S_IRWXU | S_IRWXG | S_IRWXO);
148  if (retval != 0 && errno != EEXIST) {
149  throw cms::Exception("DaqDirector")
150  << " Error creating bu run dir -: " << bu_run_dir_
151  << " mkdir error:" << strerror(errno) << "\n";
152  }
153  bu_run_open_dir_ = bu_run_dir_ + "/open";
154  retval = mkdir(bu_run_open_dir_.c_str(),
155  S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
156  if (retval != 0 && errno != EEXIST) {
157  throw cms::Exception("DaqDirector") << " Error creating bu run open dir -: "
158  << bu_run_open_dir_ << " mkdir error:" << strerror(errno)
159  << "\n";
160  }
161 
162  // the BU director does not need to know about the fu lock
163  bu_writelock_fd_ = open(bulockfile.c_str(),
164  O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
165  if (bu_writelock_fd_ == -1)
166  edm::LogWarning("EvFDaqDirector") << "problem with creating filedesc for buwritelock -: "
167  << strerror(errno);
168  else
169  edm::LogInfo("EvFDaqDirector") << "creating filedesc for buwritelock -: "
170  << bu_writelock_fd_;
171  bu_w_lock_stream = fdopen(bu_writelock_fd_, "w");
172  if (bu_w_lock_stream == 0)
173  edm::LogWarning("EvFDaqDirector")<< "Error creating write lock stream -: " << strerror(errno);
174 
175  // BU INITIALIZES LOCK FILE
176  // FU LOCK FILE OPEN
177  openFULockfileStream(fulockfile, true);
179  fflush(fu_rw_lock_stream);
180  close(fu_readwritelock_fd_);
181  }
182  else
183  {
184  // for FU, check if bu base dir exists
185 
186  retval = mkdir(bu_base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
187  if (retval != 0 && errno != EEXIST) {
188  throw cms::Exception("DaqDirector") << " Error checking for bu base dir -: "
189  << bu_base_dir_ << " mkdir error:" << strerror(errno) << "\n";
190  }
191 
193  std::string fulockfile = bu_run_dir_ + "/fu.lock";
194  openFULockfileStream(fulockfile, false);
195  }
196 
197  pthread_mutex_init(&init_lock_,NULL);
198 
199  }
struct flock bu_w_fulk
T getUntrackedParameter(std::string const &, T const &) const
struct flock fu_rw_flk
std::string run_string_
void watchPreallocate(Preallocate::slot_type const &iSlot)
void openFULockfileStream(std::string &fuLockFilePath, bool create)
void watchPreGlobalEndLumi(PreGlobalEndLumi::slot_type const &iSlot)
pthread_mutex_t init_lock_
struct flock bu_r_fulk
#define NULL
Definition: scimark2.h:8
struct flock data_rw_flk
unsigned long previousFileSize_
unsigned int jumpIndex_
struct flock fu_rw_fulk
std::string bu_base_dir_
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)
std::string bu_run_dir_
void postEndRun(edm::GlobalContext const &globalContext)
unsigned int jumpLS_
void watchPreSourceEvent(PreSourceEvent::slot_type const &iSlot)
void preallocate(edm::service::SystemBounds const &bounds)
struct flock data_rw_fulk
std::string getRunOpenDirPath() const
struct flock bu_r_flk
std::string bu_run_open_dir_
evf::EvFDaqDirector::~EvFDaqDirector ( )

Definition at line 201 of file EvFDaqDirector.cc.

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

202  {
203  if (fulocal_rwlock_fd_!=-1) {
204  unlockFULocal();
205  close(fulocal_rwlock_fd_);
206  }
207 
208  if (fulocal_rwlock_fd2_!=-1) {
209  unlockFULocal2();
210  close(fulocal_rwlock_fd2_);
211  }
212 
213  }

Member Function Documentation

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

Definition at line 52 of file EvFDaqDirector.h.

References run_dir_.

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

Definition at line 53 of file EvFDaqDirector.h.

References bu_run_dir_.

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

Definition at line 54 of file EvFDaqDirector.h.

References bu_run_open_dir_.

54 {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 
)
private

Definition at line 573 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

573  {
574 
575  if (previousFileSize_ != 0) {
576  if (!fms_) {
577  try {
579  } catch (...) {
580  edm::LogError("EvFDaqDirector") <<" FastMonitoringService not found";
581  }
582  }
584  previousFileSize_ = 0;
585  }
586 
587  struct stat buf;
588  std::stringstream ss;
589  unsigned int nextIndex = index;
590  nextIndex++;
591 
592  // 1. Check suggested file
593  nextFile = getInputJsonFilePath(ls,index);
594  if (stat(nextFile.c_str(), &buf) == 0) {
595 
596  previousFileSize_ = buf.st_size;
597  fsize = buf.st_size;
598  return true;
599  }
600  // 2. No file -> lumi ended? (and how many?)
601  else {
602  std::string BUEoLSFile = getEoLSFilePathOnBU(ls);
603  bool eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
604  unsigned int startingLumi = ls;
605  while (eolFound) {
606 
607  // recheck that no raw file appeared in the meantime
608  if (stat(nextFile.c_str(), &buf) == 0) {
609  previousFileSize_ = buf.st_size;
610  fsize = buf.st_size;
611  return true;
612  }
613 
614  int indexFilesInLS = getNFilesFromEoLS(BUEoLSFile);
615  if (indexFilesInLS < 0)
616  //parsing failed
617  return false;
618  else {
619  //check index
620  if ((int)index<indexFilesInLS) {
621  //we have 2 files, and check for 1 failed... retry (2 will never be here)
622  edm::LogError("EvFDaqDirector") << "Potential miss of index file in LS -: " << ls << ". Missing "
623  << nextFile << " because " << indexFilesInLS-1 << " is the highest index expected. Will not update fu.lock file";
624  return false;
625  }
626  }
627  // this lumi ended, check for files
628  ++ls;
629  index = 0;
630  nextFile = getInputJsonFilePath(ls,0);
631  if (stat(nextFile.c_str(), &buf) == 0) {
632  // a new file was found at new lumisection, index 0
633  previousFileSize_ = buf.st_size;
634  fsize = buf.st_size;
635 
637  // rename ended lumi to + 2
638  std::string sourceEol = getEoLSFilePathOnBU(startingLumi);
639 
640  std::string destEol = getEoLSFilePathOnBU(startingLumi+2);
641 
642  std::string cpCmd = "cp " + sourceEol + " " + destEol;
643  edm::LogInfo("EvFDaqDirector") << " testmode: Running copy cmd -: " << cpCmd;
644  int rc = system(cpCmd.c_str());
645  if (rc != 0) {
646  edm::LogError("EvFDaqDirector") << " testmode: COPY EOL FAILED!!!!! -: " << cpCmd;
647  }
648  }
649 
650  return true;
651  }
652  BUEoLSFile = getEoLSFilePathOnBU(ls);
653  eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
654  }
655  }
656  // no new file found
657  return false;
658  }
void accumulateFileSize(unsigned int lumi, unsigned long fileSize)
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::createRunOpendirMaybe ( )

Definition at line 744 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

744  {
745  // create open dir if not already there
746 
748  if (!boost::filesystem::is_directory(openPath)) {
749  LogDebug("EvFDaqDirector") << "<open> FU dir not found. Creating... -:" << openPath.string();
750  boost::filesystem::create_directories(openPath);
751  }
752  }
#define LogDebug(id)
tuple path
else: Piece not in the list, fine.
std::string getRunOpenDirPath() const
std::string evf::EvFDaqDirector::eolsFileName ( const unsigned int  ls) const
private
std::string evf::EvFDaqDirector::eorFileName ( ) const
private
std::string evf::EvFDaqDirector::findCurrentRunDir ( )
inline

Definition at line 56 of file EvFDaqDirector.h.

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

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

Definition at line 359 of file EvFDaqDirector.cc.

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

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

359  {
360  return bu_run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
361  }
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 363 of file EvFDaqDirector.cc.

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

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

363  {
364  return run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
365  }
std::string eolsFileName(const unsigned int run, const unsigned int ls)
std::string evf::EvFDaqDirector::getEoRFilePath ( ) const

Definition at line 367 of file EvFDaqDirector.cc.

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

Referenced by updateFuLock().

367  {
368  return bu_run_dir_ + "/" + fffnaming::eorFileName(run_);
369  }
std::string eorFileName(const unsigned int run)
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getEoRFilePathOnFU ( ) const

Definition at line 372 of file EvFDaqDirector.cc.

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

Referenced by FedRawDataInputSource::checkNextEvent().

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

Definition at line 331 of file EvFDaqDirector.cc.

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

Referenced by DQMFileSaver::globalBeginRun().

331  {
333  }
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 294 of file EvFDaqDirector.cc.

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

Referenced by bumpFile().

294  {
296  }
std::string inputJsonFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getJumpFilePath ( ) const
inline

Definition at line 87 of file EvFDaqDirector.h.

References bu_run_dir_, getRunNumber(), fffnaming::inputRawFileName(), jumpIndex_, and jumpLS_.

Referenced by FedRawDataInputSource::renameToNextFree().

std::string inputRawFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
unsigned int jumpIndex_
std::string bu_run_dir_
unsigned int getRunNumber() const
unsigned int jumpLS_
unsigned int evf::EvFDaqDirector::getJumpIndex ( ) const
inline

Definition at line 86 of file EvFDaqDirector.h.

References jumpIndex_.

86 { return jumpIndex_; }
unsigned int jumpIndex_
unsigned int evf::EvFDaqDirector::getJumpLS ( ) const
inline

Definition at line 85 of file EvFDaqDirector.h.

References jumpLS_.

85 { return jumpLS_; }
unsigned int jumpLS_
std::string evf::EvFDaqDirector::getMergedDatChecksumFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 327 of file EvFDaqDirector.cc.

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

327  {
329  }
std::string streamerDataChecksumFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string evf::EvFDaqDirector::getMergedDatFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 323 of file EvFDaqDirector.cc.

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

Referenced by maybeCreateAndLockFileHeadForStream().

323  {
325  }
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 343 of file EvFDaqDirector.cc.

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

343  {
345  }
std::string protocolBufferHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string evf::EvFDaqDirector::getMergedRootHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 355 of file EvFDaqDirector.cc.

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

355  {
357  }
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 512 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(), i, Json::Reader::parse(), readEolsDefinition_, matplotRender::reader, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by bumpFile().

512  {
513 
514  boost::filesystem::ifstream ij(BUEoLSFile);
515  Json::Value deserializeRoot;
517 
518  if (!reader.parse(ij, deserializeRoot)) {
519  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << BUEoLSFile;
520  return -1;
521  }
522 
524  DataPoint dp;
525  dp.deserialize(deserializeRoot);
526 
527  //read definition
528  if (readEolsDefinition_) {
529  //std::string def = boost::algorithm::trim(dp.getDefinition());
531  if (!def.size()) readEolsDefinition_=false;
532  while (def.size()) {
533  std::string fullpath;
534  if (def.find('/')==0)
535  fullpath = def;
536  else
537  fullpath = bu_run_dir_+'/'+def;
538  struct stat buf;
539  if (stat(fullpath.c_str(), &buf) == 0) {
540  DataPointDefinition eolsDpd;
541  std::string defLabel = "legend";
542  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
543  if (eolsDpd.getNames().size()==0) {
544  //try with "data" label if "legend" format is not used
545  eolsDpd = DataPointDefinition();
546  defLabel="data";
547  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
548  }
549  for (unsigned int i=0;i<eolsDpd.getNames().size();i++)
550  if (eolsDpd.getNames().at(i)=="NFiles")
552  readEolsDefinition_=false;
553  break;
554  }
555  //check if we can still find definition
556  if (def.size()<=1 || def.find('/')==std::string::npos) {
557  readEolsDefinition_=false;
558  break;
559  }
560  def = def.substr(def.find('/')+1);
561  }
562  }
563 
564  if (dp.getData().size()>eolsNFilesIndex_)
565  data = dp.getData()[eolsNFilesIndex_];
566  else {
567  edm::LogError("EvFDaqDirector") << " error reading number of files from BU JSON -: " << BUEoLSFile;
568  return -1;
569  }
570  return boost::lexical_cast<int>(data);
571  }
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:24
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 311 of file EvFDaqDirector.cc.

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

311  {
313  }
std::string streamerDataFileNameWithPid(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 307 of file EvFDaqDirector.cc.

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

307  {
309  }
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 315 of file EvFDaqDirector.cc.

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

315  {
317  }
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 335 of file EvFDaqDirector.cc.

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

335  {
337  }
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 303 of file EvFDaqDirector.cc.

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

303  {
304  return bu_run_dir_ + "/open/" + fffnaming::inputRawFileName(run_,ls,index);
305  }
std::string inputRawFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getOpenRootHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 347 of file EvFDaqDirector.cc.

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

347  {
349  }
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 319 of file EvFDaqDirector.cc.

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

319  {
321  }
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 339 of file EvFDaqDirector.cc.

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

339  {
341  }
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 299 of file EvFDaqDirector.cc.

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

Referenced by removeFile().

299  {
301  }
std::string inputRawFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
std::string bu_run_dir_
std::string evf::EvFDaqDirector::getRootHistogramFilePath ( const unsigned int  ls,
std::string const &  stream 
) const

Definition at line 351 of file EvFDaqDirector.cc.

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

351  {
353  }
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 84 of file EvFDaqDirector.h.

References run_.

Referenced by getJumpFilePath().

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

Definition at line 77 of file EvFDaqDirector.h.

References run_dir_.

Referenced by createRunOpendirMaybe(), and EvFDaqDirector().

77 {return run_dir_ +"/open";}
std::vector<int>* evf::EvFDaqDirector::getStreamFileTracker ( )
inline

Definition at line 95 of file EvFDaqDirector.h.

References streamFileTracker_.

Referenced by FedRawDataInputSource::getNextEvent().

95 {return &streamFileTracker_;}
std::vector< int > streamFileTracker_
bool evf::EvFDaqDirector::getTestModeNoBuilderUnit ( )
inline

Definition at line 88 of file EvFDaqDirector.h.

References testModeNoBuilderUnit_.

88 { return testModeNoBuilderUnit_;}
std::string evf::EvFDaqDirector::initFileName ( std::string const &  stream) const
private
std::string evf::EvFDaqDirector::inputFileNameStem ( const unsigned int  ls,
const unsigned int  index 
) const
private
bool evf::EvFDaqDirector::isSingleStreamThread ( )
inline

Definition at line 96 of file EvFDaqDirector.h.

References nStreams_, and nThreads_.

Referenced by FedRawDataInputSource::getNextEvent().

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

Definition at line 722 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

Referenced by updateFuLock().

722  {
723  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_flk);
724  flock(fulocal_rwlock_fd_,LOCK_EX);
725  }
void evf::EvFDaqDirector::lockFULocal2 ( )

Definition at line 733 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

733  {
734  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_flk2);
735  flock(fulocal_rwlock_fd2_,LOCK_EX);
736  }
void evf::EvFDaqDirector::lockInitLock ( )

Definition at line 714 of file EvFDaqDirector.cc.

References init_lock_.

714  {
715  pthread_mutex_lock(&init_lock_);
716  }
pthread_mutex_t init_lock_
FILE * evf::EvFDaqDirector::maybeCreateAndLockFileHeadForStream ( unsigned int  ls,
std::string &  stream 
)

Definition at line 694 of file EvFDaqDirector.cc.

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

694  {
695  data_rw_stream = fopen(getMergedDatFilePath(ls,stream).c_str(), "a"); //open stream for appending
697  if (data_readwrite_fd_ == -1)
698  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for datamerge "
699  << strerror(errno);
700  else
701  LogDebug("EvFDaqDirector") << "creating filedesc for datamerge -: "
703  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_flk);
704 
705  return data_rw_stream;
706  }
#define LogDebug(id)
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
void evf::EvFDaqDirector::openFULockfileStream ( std::string &  fuLockFilePath,
bool  create 
)
private

Definition at line 675 of file EvFDaqDirector.cc.

References fu_readwritelock_fd_, fu_rw_lock_stream, and LogDebug.

Referenced by EvFDaqDirector().

675  {
676  if (create) {
677  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
678  S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
679  chmod(fulockfile.c_str(),0766);
680  } else {
681  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
682  }
683  if (fu_readwritelock_fd_ == -1)
684  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
685  << " create:" << create << " error:" << strerror(errno);
686  else
687  LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: "
689 
690  fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
691  }
#define LogDebug(id)
SurfaceDeformation * create(int type, const std::vector< double > &params)
bool evf::EvFDaqDirector::outputAdler32Recheck ( ) const
inline

Definition at line 78 of file EvFDaqDirector.h.

References outputAdler32Recheck_.

std::string evf::EvFDaqDirector::outputFileNameStem ( const unsigned int  ls,
std::string const &  stream 
) const
private
void evf::EvFDaqDirector::postEndRun ( edm::GlobalContext const &  globalContext)

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

215  {
216  close(bu_readlock_fd_);
217  close(bu_writelock_fd_);
218  if (directorBu_) {
219  std::string filename = bu_run_dir_ + "/bu.lock";
220  removeFile(filename);
221  }
222  }
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 224 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

224  {
225 
226  for (unsigned int i=0;i<bounds.maxNumberOfStreams();i++){
227  streamFileTracker_.push_back(-1);
228  }
229  nThreads_=bounds.maxNumberOfStreams();
230  nStreams_=bounds.maxNumberOfThreads();
231  }
unsigned int nThreads_
int i
Definition: DBlmapReader.cc:9
std::vector< int > streamFileTracker_
unsigned int nStreams_
void evf::EvFDaqDirector::preBeginRun ( edm::GlobalContext const &  globalContext)

Definition at line 233 of file EvFDaqDirector.cc.

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

Referenced by EvFDaqDirector().

233  {
234 
235 // assert(run_ == id.run());
236 
237  // check if the requested run is the latest one - issue a warning if it isn't
239  edm::LogWarning("EvFDaqDirector") << "WARNING - checking run dir -: "
240  << run_dir_ << ". This is not the highest run "
242  }
243  }
std::string findHighestRunDir()
Definition: DirManager.cc:20
void evf::EvFDaqDirector::preGlobalEndLumi ( edm::GlobalContext const &  globalContext)

Definition at line 245 of file EvFDaqDirector.cc.

References cppFunctionSkipper::exception, fileDeleteLockPtr_, filesToDeletePtr_, LogDebug, python.rootplot.utilities::ls(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), cmsHarvester::path, and python.multivaluedict::remove().

Referenced by EvFDaqDirector().

246  {
247  //delete all files belonging to just closed lumi
248  unsigned int ls = globalContext.luminosityBlockID().luminosityBlock();
250  edm::LogWarning("EvFDaqDirector") << " Handles to check for files to delete were not set by the input source...";
251  return;
252  }
253 
254  std::unique_lock<std::mutex> lkw(*fileDeleteLockPtr_);
255  auto it = filesToDeletePtr_->begin();
256  while (it!=filesToDeletePtr_->end()) {
257  if (it->second->lumi_ == ls) {
258  const boost::filesystem::path filePath(it->second->fileName_);
259  LogDebug("EvFDaqDirector") << "Deleting input file -:" << it->second->fileName_;
260  try {
261  //rarely this fails but file gets deleted
262  boost::filesystem::remove(filePath);
263  }
264  catch (const boost::filesystem::filesystem_error& ex)
265  {
266  edm::LogError("EvFDaqDirector") << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() << ". Trying again.";
267  usleep(10000);
268  try {
269  boost::filesystem::remove(filePath);
270  }
271  catch (...) {/*file gets deleted first time but exception is still thrown*/}
272  }
273  catch (std::exception& ex)
274  {
275  edm::LogError("EvFDaqDirector") << " - deleteFile std::exception CAUGHT -: " << ex.what() << ". Trying again.";
276  usleep(10000);
277  try {
278  boost::filesystem::remove(filePath);
279  } catch (...) {/*file gets deleted first time but exception is still thrown*/}
280  }
281 
282  delete it->second;
283  it = filesToDeletePtr_->erase(it);
284  }
285  else it++;
286  }
287  }
#define LogDebug(id)
std::list< std::pair< int, InputFile * > > * filesToDeletePtr_
std::mutex * fileDeleteLockPtr_
tuple path
else: Piece not in the list, fine.
void evf::EvFDaqDirector::preSourceEvent ( edm::StreamID const &  streamID)
inline

Definition at line 289 of file EvFDaqDirector.cc.

References currentFileIndex_, and streamFileTracker_.

Referenced by EvFDaqDirector().

289  {
291  }
std::vector< int > streamFileTracker_
void evf::EvFDaqDirector::removeFile ( unsigned int  ls,
unsigned int  index 
)

Definition at line 383 of file EvFDaqDirector.cc.

References getRawFilePath().

Referenced by postEndRun().

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

376  {
377  int retval = remove(filename.c_str());
378  if (retval != 0)
379  edm::LogError("EvFDaqDirector") << "Could not remove used file -: " << filename << ". error = "
380  << strerror(errno);
381  }
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 102 of file EvFDaqDirector.h.

References fileDeleteLockPtr_, and filesToDeletePtr_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

102  {
103  fileDeleteLockPtr_=fileDeleteLock;
104  filesToDeletePtr_ = filesToDelete;
105  }
std::list< std::pair< int, InputFile * > > * filesToDeletePtr_
std::mutex * fileDeleteLockPtr_
void evf::EvFDaqDirector::setFMS ( evf::FastMonitoringService fms)
inline

Definition at line 93 of file EvFDaqDirector.h.

References fms_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

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

Definition at line 660 of file EvFDaqDirector.cc.

References fu_rw_lock_stream, getHLTprescales::readIndex(), and testModeNoBuilderUnit_.

Referenced by EvFDaqDirector().

660  {
661  if (fu_rw_lock_stream == 0)
662  edm::LogError("EvFDaqDirector") << "Error creating fu read/write lock stream "
663  << strerror(errno);
664  else {
665  edm::LogInfo("EvFDaqDirector") << "Initializing FU LOCK FILE";
666  unsigned int readLs = 1, readIndex = 0, jumpLs = 3, jumpIndex = 0;
668  fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs, readIndex,
669  jumpLs, jumpIndex);
670  else
671  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
672  }
673  }
void evf::EvFDaqDirector::unlockAndCloseMergeStream ( )

Definition at line 708 of file EvFDaqDirector.cc.

References data_readwrite_fd_, data_rw_fulk, and data_rw_stream.

708  {
709  fflush(data_rw_stream);
710  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_fulk);
711  fclose(data_rw_stream);
712  }
struct flock data_rw_fulk
void evf::EvFDaqDirector::unlockFULocal ( )

Definition at line 727 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd_.

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

727  {
728  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_fulk);
729  flock(fulocal_rwlock_fd_,LOCK_UN);
730  }
void evf::EvFDaqDirector::unlockFULocal2 ( )

Definition at line 738 of file EvFDaqDirector.cc.

References fulocal_rwlock_fd2_.

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

738  {
739  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_fulk2);
740  flock(fulocal_rwlock_fd2_,LOCK_UN);
741  }
void evf::EvFDaqDirector::unlockInitLock ( )

Definition at line 718 of file EvFDaqDirector.cc.

References init_lock_.

718  {
719  pthread_mutex_unlock(&init_lock_);
720  }
pthread_mutex_t init_lock_
void evf::EvFDaqDirector::updateFileIndex ( int const &  fileIndex)
inline

Definition at line 94 of file EvFDaqDirector.h.

References currentFileIndex_.

Referenced by FedRawDataInputSource::getNextEvent().

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

Definition at line 387 of file EvFDaqDirector.cc.

References bu_run_dir_, bumpFile(), CastorDataFrameFilter_impl::check(), fu_readwritelock_fd_, fu_rw_flk, fu_rw_fulk, fu_rw_lock_stream, getEoRFilePath(), jumpIndex_, jumpLS_, lockFULocal(), LogDebug, newFile, noFile, getHLTprescales::readIndex(), runEnded, and testModeNoBuilderUnit_.

Referenced by FedRawDataInputSource::readSupervisor().

387  {
388  EvFDaqDirector::FileStatus fileStatus = noFile;
389 
390  int retval = -1;
391  int lock_attempts = 0;
392 
393  while (retval==-1) {
394  retval = fcntl(fu_readwritelock_fd_, F_SETLK, &fu_rw_flk);
395  if (retval==-1) usleep(50000);
396  else continue;
397 
398  lock_attempts++;
399  if (lock_attempts>100 || errno==116) {
400  if (errno==116)
401  edm::LogWarning("EvFDaqDirector") << "Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
402  else
403  edm::LogWarning("EvFDaqDirector") << "Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno "<< errno <<":"<< strerror(errno) << std::endl;
404 
405  struct stat buf;
406  if (stat(bu_run_dir_.c_str(), &buf)!=0) return runEnded;
407  if (stat((bu_run_dir_+"/fu.lock").c_str(), &buf)!=0) return runEnded;
408  lock_attempts=0;
409  }
410  }
411  if(retval!=0) return fileStatus;
412 
413 #ifdef DEBUG
414  timeval ts_lockend;
415  gettimeofday(&ts_lockend,0);
416 #endif
417 
418  // if the stream is readable
419  if (fu_rw_lock_stream != 0) {
420  unsigned int readLs, readIndex, jumpLs, jumpIndex;
421  int check = 0;
422  // rewind the stream
423  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
424  // if rewinded ok
425  if (check == 0) {
426  // read its' values
428  fscanf(fu_rw_lock_stream, "%u %u %u %u", &readLs, &readIndex,
429  &jumpLs, &jumpIndex);
430  else {
431  fscanf(fu_rw_lock_stream, "%u %u", &readLs, &readIndex);
432  edm::LogInfo("EvFDaqDirector") << "Read fu.lock file file -: " << readLs << ":" << readIndex;
433  }
434 
435  // try to bump
436  bool bumpedOk = bumpFile(readLs, readIndex, nextFile, fsize);
437  ls = readLs;
438  // there is a new file to grab or lumisection ended
439  if (bumpedOk) {
440  // rewind and clear
441  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
442  if (check == 0) {
443  ftruncate(fu_readwritelock_fd_, 0);
444  fflush(fu_rw_lock_stream); //this should not be needed ???
445  } else
446  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
447  << strerror(errno);
448  // write new data
449  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
450  if (check == 0) {
451  // write next index in the file, which is the file the next process should take
453  fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs,
454  readIndex + 1, readLs + 2, readIndex + 1);
455  jumpLS_ = readLs + 2;
457  } else {
458  fprintf(fu_rw_lock_stream, "%u %u", readLs,
459  readIndex + 1);
460  }
461  fflush(fu_rw_lock_stream);
462  fsync(fu_readwritelock_fd_);
463 
464  fileStatus = newFile;
465 
467  edm::LogInfo("EvFDaqDirector") << "Written to file -: " << readLs << ":"
468  << readIndex + 1 << " --> " << readLs + 2
469  << ":" << readIndex + 1;
470  else
471  LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":"
472  << readIndex + 1;
473 
474  } else
475  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
476  << strerror(errno);
477  }
478  } else
479  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for reading failed with error "
480  << strerror(errno);
481  } else
482  edm::LogError("EvFDaqDirector") << "fu read/write lock stream is invalid " << strerror(errno);
483 
484 #ifdef DEBUG
485  timeval ts_preunlock;
486  gettimeofday(&ts_preunlock,0);
487  int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
488  double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
489 #endif
490 
491  //if new json is present, lock file which FedRawDataInputSource will later unlock
492  if (fileStatus==newFile) lockFULocal();
493 
494  //release lock at this point
495  int retvalu=-1;
496  retvalu=fcntl(fu_readwritelock_fd_, F_SETLKW, &fu_rw_fulk);
497  if (retvalu==-1) edm::LogError("EvFDaqDirector") << "Error unlocking the fu.lock " << strerror(errno);
498 
499 #ifdef DEBUG
500  edm::LogDebug("EvFDaqDirector") << "Waited during lock -: " << locked_period << " seconds";
501 #endif
502 
503  if ( fileStatus == noFile ) {
504  struct stat buf;
505  //edm::LogInfo("EvFDaqDirector") << " looking for EoR file: " << getEoRFilePath().c_str();
506  if ( stat(getEoRFilePath().c_str(), &buf) == 0 || stat(bu_run_dir_.c_str(), &buf)!=0)
507  fileStatus = runEnded;
508  }
509  return fileStatus;
510  }
#define LogDebug(id)
struct flock fu_rw_flk
std::string getEoRFilePath() const
unsigned int jumpIndex_
struct flock fu_rw_fulk
bool check(const DataFrame &df, bool capcheck, bool dvercheck)
bool bumpFile(unsigned int &ls, unsigned int &index, std::string &nextFile, uint32_t &fsize)
std::string bu_run_dir_
unsigned int jumpLS_

Member Data Documentation

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

Definition at line 122 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

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

Definition at line 123 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

struct flock evf::EvFDaqDirector::bu_r_flk
private

Definition at line 154 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_r_fulk
private

Definition at line 156 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_r_lock_stream
private

Definition at line 142 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_readlock_fd_
private

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

Referenced by buBaseRunOpenDir(), and EvFDaqDirector().

FILE* evf::EvFDaqDirector::bu_t_monitor_stream
private

Definition at line 145 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_flk
private

Definition at line 153 of file EvFDaqDirector.h.

struct flock evf::EvFDaqDirector::bu_w_fulk
private

Definition at line 155 of file EvFDaqDirector.h.

FILE* evf::EvFDaqDirector::bu_w_lock_stream
private

Definition at line 141 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

FILE* evf::EvFDaqDirector::bu_w_monitor_stream
private

Definition at line 144 of file EvFDaqDirector.h.

int evf::EvFDaqDirector::bu_writelock_fd_
private

Definition at line 135 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and postEndRun().

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

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

Referenced by maybeCreateAndLockFileHeadForStream().

struct flock evf::EvFDaqDirector::data_rw_fulk
private

Definition at line 160 of file EvFDaqDirector.h.

Referenced by unlockAndCloseMergeStream().

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

Definition at line 124 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector(), and postEndRun().

DirManager evf::EvFDaqDirector::dirManager_
private

Definition at line 148 of file EvFDaqDirector.h.

Referenced by findCurrentRunDir(), and preBeginRun().

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

Definition at line 179 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

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

Definition at line 170 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

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

Definition at line 171 of file EvFDaqDirector.h.

Referenced by preGlobalEndLumi(), and setDeleteTracking().

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

Definition at line 166 of file EvFDaqDirector.h.

Referenced by bumpFile(), and setFMS().

int evf::EvFDaqDirector::fu_readwritelock_fd_
private

Definition at line 136 of file EvFDaqDirector.h.

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

struct flock evf::EvFDaqDirector::fu_rw_flk
private

Definition at line 157 of file EvFDaqDirector.h.

Referenced by updateFuLock().

struct flock evf::EvFDaqDirector::fu_rw_fulk
private

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

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

int evf::EvFDaqDirector::fulocal_rwlock_fd_
private

Definition at line 138 of file EvFDaqDirector.h.

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

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

Definition at line 173 of file EvFDaqDirector.h.

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

unsigned int evf::EvFDaqDirector::jumpIndex_
private

Definition at line 151 of file EvFDaqDirector.h.

Referenced by getJumpFilePath(), getJumpIndex(), and updateFuLock().

unsigned int evf::EvFDaqDirector::jumpLS_
private

Definition at line 151 of file EvFDaqDirector.h.

Referenced by getJumpFilePath(), getJumpLS(), and updateFuLock().

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

Definition at line 175 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

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

Definition at line 176 of file EvFDaqDirector.h.

Referenced by isSingleStreamThread(), and preallocate().

bool evf::EvFDaqDirector::outputAdler32Recheck_
private

Definition at line 126 of file EvFDaqDirector.h.

Referenced by outputAdler32Recheck().

unsigned long evf::EvFDaqDirector::previousFileSize_
private

Definition at line 150 of file EvFDaqDirector.h.

Referenced by bumpFile().

bool evf::EvFDaqDirector::readEolsDefinition_ = true
private

Definition at line 178 of file EvFDaqDirector.h.

Referenced by getNFilesFromEoLS().

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

Definition at line 129 of file EvFDaqDirector.h.

Referenced by EvFDaqDirector().

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

Definition at line 167 of file EvFDaqDirector.h.

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

bool evf::EvFDaqDirector::testModeNoBuilderUnit_
private