CMS 3D CMS Logo

All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Pages
EvFDaqDirector.cc
Go to the documentation of this file.
10 
16 
17 #include <iostream>
18 #include <sstream>
19 #include <sys/time.h>
20 #include <unistd.h>
21 #include <cstdio>
22 #include <boost/lexical_cast.hpp>
23 #include <boost/filesystem/fstream.hpp>
24 
25 //#define DEBUG
26 
27 using namespace jsoncollector;
28 
29 
30 namespace evf {
31 
32  //for enum MergeType
33  const std::vector<std::string> EvFDaqDirector::MergeTypeNames_ = {"","DAT","PB","JSNDATA"};
34 
35  EvFDaqDirector::EvFDaqDirector(const edm::ParameterSet &pset,
36  edm::ActivityRegistry& reg) :
37  base_dir_(pset.getUntrackedParameter<std::string> ("baseDir", ".")),
38  bu_base_dir_(pset.getUntrackedParameter<std::string> ("buBaseDir", ".")),
39  directorBu_(pset.getUntrackedParameter<bool> ("directorIsBu", false)),
40  run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
41  outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck",false)),
42  requireTSPSet_(pset.getUntrackedParameter<bool>("requireTransfersPSet",false)),
43  selectedTransferMode_(pset.getUntrackedParameter<std::string>("selectedTransferMode","")),
44  hltSourceDirectory_(pset.getUntrackedParameter<std::string>("hltSourceDirectory","")),
45  fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>("fuLockPollInterval",2000)),
46  mergeTypePset_(pset.getUntrackedParameter<std::string>("mergeTypePset","")),
47  hostname_(""),
48  bu_readlock_fd_(-1),
49  bu_writelock_fd_(-1),
50  fu_readwritelock_fd_(-1),
51  fulocal_rwlock_fd_(-1),
52  fulocal_rwlock_fd2_(-1),
53  bu_w_lock_stream(nullptr),
54  bu_r_lock_stream(nullptr),
55  fu_rw_lock_stream(nullptr),
56  dirManager_(base_dir_),
57  previousFileSize_(0),
58  bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
59  bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
60  bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
61  bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
62  fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
63  fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
64  {
65 
71 
72  //save hostname for later
73  char hostname[33];
74  gethostname(hostname,32);
75  hostname_ = hostname;
76 
77  char * fuLockPollIntervalPtr = getenv("FFF_LOCKPOLLINTERVAL");
78  if (fuLockPollIntervalPtr) {
79  try {
80  fuLockPollInterval_=boost::lexical_cast<unsigned int>(std::string(fuLockPollIntervalPtr));
81  edm::LogInfo("EvFDaqDirector") << "Setting fu lock poll interval by environment string: " << fuLockPollInterval_ << " us";
82  }
83  catch( boost::bad_lexical_cast const& ) {
84  edm::LogWarning("EvFDaqDirector") << "Bad lexical cast in parsing: " << std::string(fuLockPollIntervalPtr);
85  }
86  }
87 
88  }
89 
91  {
92  std::stringstream ss;
93  ss << "run" << std::setfill('0') << std::setw(6) << run_;
94  run_string_ = ss.str();
96 
97  // check if base dir exists or create it accordingly
98  int retval = mkdir(base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
99  if (retval != 0 && errno != EEXIST) {
100  throw cms::Exception("DaqDirector") << " Error checking for base dir -: "
101  << base_dir_ << " mkdir error:" << strerror(errno);
102  }
103 
104  //create run dir in base dir
105  umask(0);
106  retval = mkdir(run_dir_.c_str(),
107  S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
108  if (retval != 0 && errno != EEXIST) {
109  throw cms::Exception("DaqDirector") << " Error creating run dir -: "
110  << run_dir_ << " mkdir error:" << strerror(errno);
111  }
112 
113  //create fu-local.lock in run open dir
114  if (!directorBu_) {
115 
117  std::string fulocal_lock_ = getRunOpenDirPath() +"/fu-local.lock";
118  fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
119  if (fulocal_rwlock_fd_==-1)
120  throw cms::Exception("DaqDirector") << " Error creating/opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
121  chmod(fulocal_lock_.c_str(),0777);
122  fsync(fulocal_rwlock_fd_);
123  //open second fd for another input source thread
124  fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
125  if (fulocal_rwlock_fd2_==-1)
126  throw cms::Exception("DaqDirector") << " Error opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
127  }
128 
129  //bu_run_dir: for FU, for which the base dir is local and the BU is remote, it is expected to be there
130  //for BU, it is created at this point
131  if (directorBu_)
132  {
134  std::string bulockfile = bu_run_dir_ + "/bu.lock";
135  std::string fulockfile = bu_run_dir_ + "/fu.lock";
136 
137  //make or find bu run dir
138  retval = mkdir(bu_run_dir_.c_str(),
139  S_IRWXU | S_IRWXG | S_IRWXO);
140  if (retval != 0 && errno != EEXIST) {
141  throw cms::Exception("DaqDirector")
142  << " Error creating bu run dir -: " << bu_run_dir_
143  << " mkdir error:" << strerror(errno) << "\n";
144  }
145  bu_run_open_dir_ = bu_run_dir_ + "/open";
146  retval = mkdir(bu_run_open_dir_.c_str(),
147  S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
148  if (retval != 0 && errno != EEXIST) {
149  throw cms::Exception("DaqDirector") << " Error creating bu run open dir -: "
150  << bu_run_open_dir_ << " mkdir error:" << strerror(errno)
151  << "\n";
152  }
153 
154  // the BU director does not need to know about the fu lock
155  bu_writelock_fd_ = open(bulockfile.c_str(),
156  O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
157  if (bu_writelock_fd_ == -1)
158  edm::LogWarning("EvFDaqDirector") << "problem with creating filedesc for buwritelock -: "
159  << strerror(errno);
160  else
161  edm::LogInfo("EvFDaqDirector") << "creating filedesc for buwritelock -: "
162  << bu_writelock_fd_;
163  bu_w_lock_stream = fdopen(bu_writelock_fd_, "w");
164  if (bu_w_lock_stream == nullptr)
165  edm::LogWarning("EvFDaqDirector")<< "Error creating write lock stream -: " << strerror(errno);
166 
167  // BU INITIALIZES LOCK FILE
168  // FU LOCK FILE OPEN
169  openFULockfileStream(fulockfile, true);
171  fflush(fu_rw_lock_stream);
172  close(fu_readwritelock_fd_);
173 
174  if (!hltSourceDirectory_.empty())
175  {
176  struct stat buf;
177  if (stat(hltSourceDirectory_.c_str(),&buf)==0) {
178  std::string hltdir=bu_run_dir_+"/hlt";
179  std::string tmphltdir=bu_run_open_dir_+"/hlt";
180  retval = mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
181  if (retval != 0 && errno != EEXIST)
182  throw cms::Exception("DaqDirector")
183  << " Error creating bu run dir -: " << hltdir
184  << " mkdir error:" << strerror(errno) << "\n";
185 
186  boost::filesystem::copy_file(hltSourceDirectory_+"/HltConfig.py",tmphltdir+"/HltConfig.py");
187 
188  boost::filesystem::copy_file(hltSourceDirectory_+"/fffParameters.jsn",tmphltdir+"/fffParameters.jsn");
189 
190  boost::filesystem::rename(tmphltdir,hltdir);
191  }
192  else
193  throw cms::Exception("DaqDirector") << " Error looking for HLT configuration -: " << hltSourceDirectory_;
194  }
195  //else{}//no configuration specified
196  }
197  else
198  {
199  // for FU, check if bu base dir exists
200 
201  retval = mkdir(bu_base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
202  if (retval != 0 && errno != EEXIST) {
203  throw cms::Exception("DaqDirector") << " Error checking for bu base dir -: "
204  << bu_base_dir_ << " mkdir error:" << strerror(errno) << "\n";
205  }
206 
208  std::string fulockfile = bu_run_dir_ + "/fu.lock";
209  openFULockfileStream(fulockfile, false);
210  }
211 
212  pthread_mutex_init(&init_lock_,nullptr);
213 
214  stopFilePath_ = run_dir_+"/CMSSW_STOP";
215  std::stringstream sstp;
216  sstp << stopFilePath_ << "_pid" << getpid();
217  stopFilePathPid_ = sstp.str();
218  }
219 
221  {
222  if (fulocal_rwlock_fd_!=-1) {
223  unlockFULocal();
224  close(fulocal_rwlock_fd_);
225  }
226 
227  if (fulocal_rwlock_fd2_!=-1) {
228  unlockFULocal2();
229  close(fulocal_rwlock_fd2_);
230  }
231 
232  }
233 
234 
236 
237  initRun();
238 
239  nThreads_=bounds.maxNumberOfStreams();
240  nStreams_=bounds.maxNumberOfThreads();
241  }
242 
244  {
246  desc.setComment("Service used for file locking arbitration and for propagating information between other EvF components");
247  desc.addUntracked<std::string> ("baseDir", ".")->setComment("Local base directory for run output");
248  desc.addUntracked<std::string> ("buBaseDir", ".")->setComment("BU base ramdisk directory ");
249  desc.addUntracked<unsigned int> ("runNumber",0)->setComment("Run Number in ramdisk to open");
250  desc.addUntracked<bool>("outputAdler32Recheck",false)->setComment("Check Adler32 of per-process output files while micro-merging");
251  desc.addUntracked<bool>("requireTransfersPSet",false)->setComment("Require complete transferSystem PSet in the process configuration");
252  desc.addUntracked<std::string>("selectedTransferMode","")->setComment("Selected transfer mode (choice in Lvl0 propagated as Python parameter");
253  desc.addUntracked<unsigned int>("fuLockPollInterval",2000)->setComment("Lock polling interval in microseconds for the input directory file lock");
254  desc.addUntracked<std::string>("mergingPset","")->setComment("Name of merging PSet to look for merging type definitions for streams");
255  desc.setAllowAnything();
256  descriptions.add("EvFDaqDirector", desc);
257  }
258 
260  edm::ProcessContext const& pc) {
262  checkMergeTypePSet(pc);
263  }
264 
265  void EvFDaqDirector::preBeginRun(edm::GlobalContext const& globalContext) {
266 
267  //assert(run_ == id.run());
268 
269  // check if the requested run is the latest one - issue a warning if it isn't
271  edm::LogWarning("EvFDaqDirector") << "WARNING - checking run dir -: "
272  << run_dir_ << ". This is not the highest run "
274  }
275  }
276 
277  void EvFDaqDirector::postEndRun(edm::GlobalContext const& globalContext) {
278  close(bu_readlock_fd_);
279  close(bu_writelock_fd_);
280  if (directorBu_) {
281  std::string filename = bu_run_dir_ + "/bu.lock";
282  removeFile(filename);
283  }
284  }
285 
287  {
288  //delete all files belonging to just closed lumi
289  unsigned int ls = globalContext.luminosityBlockID().luminosityBlock();
291  edm::LogWarning("EvFDaqDirector") << " Handles to check for files to delete were not set by the input source...";
292  return;
293  }
294 
295  std::unique_lock<std::mutex> lkw(*fileDeleteLockPtr_);
296  auto it = filesToDeletePtr_->begin();
297  while (it!=filesToDeletePtr_->end()) {
298  if (it->second->lumi_ == ls) {
299  const boost::filesystem::path filePath(it->second->fileName_);
300  LogDebug("EvFDaqDirector") << "Deleting input file -:" << it->second->fileName_;
301  try {
302  //rarely this fails but file gets deleted
303  boost::filesystem::remove(filePath);
304  }
305  catch (const boost::filesystem::filesystem_error& ex)
306  {
307  edm::LogError("EvFDaqDirector") << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() << ". Trying again.";
308  usleep(10000);
309  try {
310  boost::filesystem::remove(filePath);
311  }
312  catch (const boost::filesystem::filesystem_error&) {/*file gets deleted first time but exception is still thrown*/}
313  }
314  catch (std::exception& ex)
315  {
316  edm::LogError("EvFDaqDirector") << " - deleteFile std::exception CAUGHT -: " << ex.what() << ". Trying again.";
317  usleep(10000);
318  try {
319  boost::filesystem::remove(filePath);
320  } catch (std::exception&) {/*file gets deleted first time but exception is still thrown*/}
321  }
322 
323  delete it->second;
324  it = filesToDeletePtr_->erase(it);
325  }
326  else it++;
327  }
328  }
329 
330  std::string EvFDaqDirector::getInputJsonFilePath(const unsigned int ls, const unsigned int index) const {
331  return bu_run_dir_ + "/" + fffnaming::inputJsonFileName(run_,ls,index);
332  }
333 
334  std::string EvFDaqDirector::getRawFilePath(const unsigned int ls, const unsigned int index) const {
335  return bu_run_dir_ + "/" + fffnaming::inputRawFileName(run_,ls,index);
336  }
337 
338  std::string EvFDaqDirector::getOpenRawFilePath(const unsigned int ls, const unsigned int index) const {
339  return bu_run_dir_ + "/open/" + fffnaming::inputRawFileName(run_,ls,index);
340  }
341 
342  std::string EvFDaqDirector::getOpenInputJsonFilePath(const unsigned int ls, const unsigned int index) const {
343  return bu_run_dir_ + "/open/" + fffnaming::inputJsonFileName(run_,ls,index);
344  }
345 
346  std::string EvFDaqDirector::getDatFilePath(const unsigned int ls, std::string const& stream) const {
347  return run_dir_ + "/" + fffnaming::streamerDataFileNameWithPid(run_,ls,stream);
348  }
349 
350  std::string EvFDaqDirector::getOpenDatFilePath(const unsigned int ls, std::string const& stream) const {
351  return run_dir_ + "/open/" + fffnaming::streamerDataFileNameWithPid(run_,ls,stream);
352  }
353 
354  std::string EvFDaqDirector::getOpenOutputJsonFilePath(const unsigned int ls, std::string const& stream) const {
355  return run_dir_ + "/open/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
356  }
357 
358  std::string EvFDaqDirector::getOutputJsonFilePath(const unsigned int ls, std::string const& stream) const {
359  return run_dir_ + "/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
360  }
361 
362  std::string EvFDaqDirector::getMergedDatFilePath(const unsigned int ls, std::string const& stream) const {
364  }
365 
366  std::string EvFDaqDirector::getMergedDatChecksumFilePath(const unsigned int ls, std::string const& stream) const {
368  }
369 
371  return run_dir_ + "/open/" + fffnaming::initFileNameWithPid(run_,0,stream);
372  }
373 
375  return run_dir_ + "/" + fffnaming::initFileNameWithPid(run_,0,stream);
376  }
377 
379  return run_dir_ + "/open/" + fffnaming::protocolBufferHistogramFileNameWithPid(run_,ls,stream);
380  }
381 
384  }
385 
388  }
389 
390  std::string EvFDaqDirector::getOpenRootHistogramFilePath(const unsigned int ls, std::string const& stream) const {
391  return run_dir_ + "/open/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
392  }
393 
394  std::string EvFDaqDirector::getRootHistogramFilePath(const unsigned int ls, std::string const& stream) const {
395  return run_dir_ + "/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
396  }
397 
400  }
401 
403  return bu_run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
404  }
405 
407  return run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
408  }
409 
411  return run_dir_ + "/" + fffnaming::bolsFileName(run_,ls);
412  }
413 
415  return bu_run_dir_ + "/" + fffnaming::eorFileName(run_);
416  }
417 
418 
420  return run_dir_ + "/" + fffnaming::eorFileName(run_);
421  }
422 
424  int retval = remove(filename.c_str());
425  if (retval != 0)
426  edm::LogError("EvFDaqDirector") << "Could not remove used file -: " << filename << ". error = "
427  << strerror(errno);
428  }
429 
430  void EvFDaqDirector::removeFile(unsigned int ls, unsigned int index) {
431  removeFile(getRawFilePath(ls,index));
432  }
433 
434  EvFDaqDirector::FileStatus EvFDaqDirector::updateFuLock(unsigned int& ls, std::string& nextFile, uint32_t& fsize, uint64_t& lockWaitTime) {
435  EvFDaqDirector::FileStatus fileStatus = noFile;
436 
437  int retval = -1;
438  int lock_attempts = 0;
439 
440  struct stat buf;
441  int stopFileLS = -1;
442  int stopFileCheck = stat(stopFilePath_.c_str(),&buf);
443  int stopFilePidCheck = stat(stopFilePathPid_.c_str(),&buf);
444  if (stopFileCheck==0 || stopFilePidCheck==0) {
445  if (stopFileCheck==0)
446  stopFileLS = readLastLSEntry(stopFilePath_);
447  else
448  stopFileLS = 1;//stop without drain if only pid is stopped
449  if (!stop_ls_override_) {
450  //if lumisection is higher than in stop file, should quit at next from current
451  if (stopFileLS>=0 && (int)ls>=stopFileLS) stopFileLS = stop_ls_override_ = ls;
452  }
453  else stopFileLS = stop_ls_override_;
454  edm::LogWarning("EvFDaqDirector") << "Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
455  //return runEnded;
456  }
457  else //if file was removed before reaching stop condition, reset this
458  stop_ls_override_ = 0;
459 
460  timeval ts_lockbegin;
461  gettimeofday(&ts_lockbegin,nullptr);
462  long total_lock_attempts = 0;
463 
464  while (retval==-1) {
465  retval = fcntl(fu_readwritelock_fd_, F_SETLK, &fu_rw_flk);
466  if (retval==-1) usleep(fuLockPollInterval_);
467  else continue;
468 
469  lock_attempts+=fuLockPollInterval_;
470  total_lock_attempts+=fuLockPollInterval_;
471  if (lock_attempts>5000000 || errno==116) {
472  if (errno==116)
473  edm::LogWarning("EvFDaqDirector") << "Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
474  else
475  edm::LogWarning("EvFDaqDirector") << "Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno "
476  << errno <<":"<< strerror(errno) << std::endl;
477 
478 
479  if (stat(getEoLSFilePathOnFU(ls).c_str(),&buf)==0) {
480  edm::LogWarning("EvFDaqDirector") << "Detected local EoLS for lumisection "<< ls ;
481  ls++;
482  return noFile;
483  }
484 
485  if (stat(bu_run_dir_.c_str(), &buf)!=0) return runEnded;
486  if (stat((bu_run_dir_+"/fu.lock").c_str(), &buf)!=0) return runEnded;
487  lock_attempts=0;
488  }
489  if (total_lock_attempts>5*60000000) {
490  edm::LogError("EvFDaqDirector") << "Unable to obtain a lock for 5 minutes. Stopping polling activity.";
491  return runAbort;
492  }
493  }
494 
495  timeval ts_lockend;
496  gettimeofday(&ts_lockend,nullptr);
497  long deltat = (ts_lockend.tv_usec-ts_lockbegin.tv_usec) + (ts_lockend.tv_sec-ts_lockbegin.tv_sec)*1000000;
498  if (deltat>0.) lockWaitTime=deltat;
499 
500 
501 
502  if(retval!=0) return fileStatus;
503 
504 #ifdef DEBUG
505  timeval ts_lockend;
506  gettimeofday(&ts_lockend,0);
507 #endif
508 
509  // if the stream is readable
510  if (fu_rw_lock_stream != nullptr) {
511  unsigned int readLs, readIndex;
512  int check = 0;
513  // rewind the stream
514  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
515  // if rewinded ok
516  if (check == 0) {
517  // read its' values
518  fscanf(fu_rw_lock_stream, "%u %u", &readLs, &readIndex);
519  edm::LogInfo("EvFDaqDirector") << "Read fu.lock file file -: " << readLs << ":" << readIndex;
520 
521  unsigned int currentLs = readLs;
522  bool bumpedOk = false;
523  //if next lumisection in a lock file is not +1 wrt. source, cycle through the next empty one, unless initial lumi not yet set
524  //no lock file write in this case
525  if (ls && ls+1 < currentLs) ls++;
526  else {
527  // try to bump (look for new index or EoLS file)
528  bumpedOk = bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
529  //avoid 2 lumisections jump
530  if (ls && readLs>currentLs && currentLs > ls) {
531  ls++;
532  readLs=currentLs=ls;
533  readIndex=0;
534  bumpedOk=false;
535  //no write to lock file
536  }
537  else {
538  if (ls==0 && readLs>currentLs) {
539  //make sure to intialize always with LS found in the lock file, with possibility of grabbing index file immediately
540  //in this case there is no new file in the same LS
541  readLs=currentLs;
542  readIndex=0;
543  bumpedOk=false;
544  //no write to lock file
545  }
546  //update return LS value
547  ls = readLs;
548  }
549  }
550  if (bumpedOk) {
551  // there is a new index file to grab, lock file needs to be updated
552  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
553  if (check == 0) {
554  ftruncate(fu_readwritelock_fd_, 0);
555  // write next index in the file, which is the file the next process should take
556  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex + 1);
557  fflush(fu_rw_lock_stream);
558  fsync(fu_readwritelock_fd_);
559  fileStatus = newFile;
560  LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":" << readIndex + 1;
561  }
562  else {
563  throw cms::Exception("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error " << strerror(errno);
564  }
565  }
566  else if (currentLs < readLs) {
567  //there is no new file in next LS (yet), but lock file can be updated to the next LS
568  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
569  if (check == 0) {
570  ftruncate(fu_readwritelock_fd_, 0);
571  // in this case LS was bumped, but no new file. Thus readIndex is 0 (set by bumpFile)
572  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
573  fflush(fu_rw_lock_stream);
574  fsync(fu_readwritelock_fd_);
575  LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":" << readIndex;
576  }
577  else {
578  throw cms::Exception("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error " << strerror(errno);
579  }
580  }
581  } else {
582  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for reading failed with error " << strerror(errno);
583  }
584  } else {
585  edm::LogError("EvFDaqDirector") << "fu read/write lock stream is invalid " << strerror(errno);
586  }
587 
588 #ifdef DEBUG
589  timeval ts_preunlock;
590  gettimeofday(&ts_preunlock,0);
591  int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
592  double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
593 #endif
594 
595  //if new json is present, lock file which FedRawDataInputSource will later unlock
596  if (fileStatus==newFile) lockFULocal();
597 
598  //release lock at this point
599  int retvalu=-1;
600  retvalu=fcntl(fu_readwritelock_fd_, F_SETLKW, &fu_rw_fulk);
601  if (retvalu==-1) edm::LogError("EvFDaqDirector") << "Error unlocking the fu.lock " << strerror(errno);
602 
603 #ifdef DEBUG
604  edm::LogDebug("EvFDaqDirector") << "Waited during lock -: " << locked_period << " seconds";
605 #endif
606 
607  if ( fileStatus == noFile ) {
608  struct stat buf;
609  //edm::LogInfo("EvFDaqDirector") << " looking for EoR file: " << getEoRFilePath().c_str();
610  if ( stat(getEoRFilePath().c_str(), &buf) == 0 || stat(bu_run_dir_.c_str(), &buf)!=0)
611  fileStatus = runEnded;
612  if (stopFileLS>=0 && (int)ls > stopFileLS) {
613  edm::LogInfo("EvFDaqDirector") << "Reached maximum lumisection set by hltd";
614  fileStatus = runEnded;
615  }
616  }
617  return fileStatus;
618  }
619 
621 
622  boost::filesystem::ifstream ij(BUEoLSFile);
623  Json::Value deserializeRoot;
625 
626  if (!reader.parse(ij, deserializeRoot)) {
627  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << BUEoLSFile;
628  return -1;
629  }
630 
632  DataPoint dp;
633  dp.deserialize(deserializeRoot);
634 
635  //read definition
636  if (readEolsDefinition_) {
637  //std::string def = boost::algorithm::trim(dp.getDefinition());
639  if (def.empty()) readEolsDefinition_=false;
640  while (!def.empty()) {
641  std::string fullpath;
642  if (def.find('/')==0)
643  fullpath = def;
644  else
645  fullpath = bu_run_dir_+'/'+def;
646  struct stat buf;
647  if (stat(fullpath.c_str(), &buf) == 0) {
648  DataPointDefinition eolsDpd;
649  std::string defLabel = "legend";
650  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
651  if (eolsDpd.getNames().empty()) {
652  //try with "data" label if "legend" format is not used
653  eolsDpd = DataPointDefinition();
654  defLabel="data";
655  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
656  }
657  for (unsigned int i=0;i<eolsDpd.getNames().size();i++)
658  if (eolsDpd.getNames().at(i)=="NFiles")
660  readEolsDefinition_=false;
661  break;
662  }
663  //check if we can still find definition
664  if (def.size()<=1 || def.find('/')==std::string::npos) {
665  readEolsDefinition_=false;
666  break;
667  }
668  def = def.substr(def.find('/')+1);
669  }
670  }
671 
672  if (dp.getData().size()>eolsNFilesIndex_)
673  data = dp.getData()[eolsNFilesIndex_];
674  else {
675  edm::LogError("EvFDaqDirector") << " error reading number of files from BU JSON -: " << BUEoLSFile;
676  return -1;
677  }
678  return boost::lexical_cast<int>(data);
679  }
680 
681  bool EvFDaqDirector::bumpFile(unsigned int& ls, unsigned int& index, std::string& nextFile, uint32_t& fsize, int maxLS) {
682 
683  if (previousFileSize_ != 0) {
684  if (!fms_) {
686  }
688  previousFileSize_ = 0;
689  }
690 
691  //reached limit
692  if (maxLS>=0 && ls > (unsigned int)maxLS) return false;
693 
694  struct stat buf;
695  std::stringstream ss;
696  unsigned int nextIndex = index;
697  nextIndex++;
698 
699  // 1. Check suggested file
700  nextFile = getInputJsonFilePath(ls,index);
701  if (stat(nextFile.c_str(), &buf) == 0) {
702 
703  previousFileSize_ = buf.st_size;
704  fsize = buf.st_size;
705  return true;
706  }
707  // 2. No file -> lumi ended? (and how many?)
708  else {
709  std::string BUEoLSFile = getEoLSFilePathOnBU(ls);
710  bool eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
711  while (eolFound) {
712 
713  // recheck that no raw file appeared in the meantime
714  if (stat(nextFile.c_str(), &buf) == 0) {
715  previousFileSize_ = buf.st_size;
716  fsize = buf.st_size;
717  return true;
718  }
719 
720  int indexFilesInLS = getNFilesFromEoLS(BUEoLSFile);
721  if (indexFilesInLS < 0)
722  //parsing failed
723  return false;
724  else {
725  //check index
726  if ((int)index<indexFilesInLS) {
727  //we have 2 files, and check for 1 failed... retry (2 will never be here)
728  edm::LogError("EvFDaqDirector") << "Potential miss of index file in LS -: " << ls << ". Missing "
729  << nextFile << " because " << indexFilesInLS-1 << " is the highest index expected. Will not update fu.lock file";
730  return false;
731  }
732  }
733  // this lumi ended, check for files
734  ++ls;
735  index = 0;
736 
737  //reached limit
738  if (maxLS>=0 && ls > (unsigned int)maxLS) return false;
739 
740  nextFile = getInputJsonFilePath(ls,0);
741  if (stat(nextFile.c_str(), &buf) == 0) {
742  // a new file was found at new lumisection, index 0
743  previousFileSize_ = buf.st_size;
744  fsize = buf.st_size;
745  return true;
746  }
747  else {
748  //change of policy: we need to cycle through each LS
749  return false;
750  }
751  BUEoLSFile = getEoLSFilePathOnBU(ls);
752  eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
753  }
754  }
755  // no new file found
756  return false;
757  }
758 
760  if (fu_rw_lock_stream == nullptr)
761  edm::LogError("EvFDaqDirector") << "Error creating fu read/write lock stream "
762  << strerror(errno);
763  else {
764  edm::LogInfo("EvFDaqDirector") << "Initializing FU LOCK FILE";
765  unsigned int readLs = 1, readIndex = 0;
766  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
767  }
768  }
769 
771  if (create) {
772  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
773  S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
774  chmod(fulockfile.c_str(),0766);
775  } else {
776  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
777  }
778  if (fu_readwritelock_fd_ == -1)
779  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
780  << " create:" << create << " error:" << strerror(errno);
781  else
782  LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: "
784 
785  fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
786  }
787 
789  pthread_mutex_lock(&init_lock_);
790  }
791 
793  pthread_mutex_unlock(&init_lock_);
794  }
795 
797  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_flk);
798  flock(fulocal_rwlock_fd_,LOCK_SH);
799  }
800 
802  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_fulk);
803  flock(fulocal_rwlock_fd_,LOCK_UN);
804  }
805 
806 
808  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_flk2);
809  flock(fulocal_rwlock_fd2_,LOCK_EX);
810  }
811 
813  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_fulk2);
814  flock(fulocal_rwlock_fd2_,LOCK_UN);
815  }
816 
817 
819  // create open dir if not already there
820 
822  if (!boost::filesystem::is_directory(openPath)) {
823  LogDebug("EvFDaqDirector") << "<open> FU dir not found. Creating... -:" << openPath.string();
824  boost::filesystem::create_directories(openPath);
825  }
826  }
827 
828 
830 
831  boost::filesystem::ifstream ij(file);
832  Json::Value deserializeRoot;
834 
835  if (!reader.parse(ij, deserializeRoot)) {
836  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << file;
837  return -1;
838  }
839 
840  int ret = deserializeRoot.get("lastLS","").asInt();
841  return ret;
842 
843  }
844 
845  //if transferSystem PSet is present in the menu, we require it to be complete and consistent for all specified streams
847  {
848  if(transferSystemJson_) return;
849 
850  transferSystemJson_.reset(new Json::Value);
852  if (topPset.existsAs<edm::ParameterSet>("transferSystem",true))
853  {
854  const edm::ParameterSet& tsPset(topPset.getParameterSet("transferSystem"));
855 
856  Json::Value destinationsVal(Json::arrayValue);
857  std::vector<std::string> destinations = tsPset.getParameter<std::vector<std::string>>("destinations");
858  for (auto & dest: destinations) destinationsVal.append(dest);
859  (*transferSystemJson_)["destinations"]=destinationsVal;
860 
861  Json::Value modesVal(Json::arrayValue);
862  std::vector<std::string> modes = tsPset.getParameter< std::vector<std::string> >("transferModes");
863  for (auto & mode: modes) modesVal.append(mode);
864  (*transferSystemJson_)["transferModes"]=modesVal;
865 
866  for (auto psKeyItr =tsPset.psetTable().begin();psKeyItr!=tsPset.psetTable().end(); ++ psKeyItr) {
867  if (psKeyItr->first!="destinations" && psKeyItr->first!="transferModes") {
868  const edm::ParameterSet & streamDef = tsPset.getParameterSet(psKeyItr->first);
869  Json::Value streamVal;
870  for (auto & mode : modes) {
871  //validation
872  if (!streamDef.existsAs<std::vector<std::string>>(mode,true))
873  throw cms::Exception("EvFDaqDirector") << " Missing transfer system specification for -:" << psKeyItr->first << " (transferMode " << mode << ")";
874  std::vector<std::string> streamDestinations = streamDef.getParameter<std::vector<std::string>>(mode);
875 
876  Json::Value sDestsValue(Json::arrayValue);
877 
878  if (streamDestinations.empty())
879  throw cms::Exception("EvFDaqDirector") << " Missing transter system destination(s) for -: "<< psKeyItr->first << ", mode:" << mode;
880 
881  for (auto & sdest:streamDestinations) {
882  bool sDestValid=false;
883  sDestsValue.append(sdest);
884  for (auto & dest: destinations) {
885  if (dest==sdest) sDestValid=true;
886  }
887  if (!sDestValid)
888  throw cms::Exception("EvFDaqDirector") << " Invalid transter system destination specified for -: "<< psKeyItr->first << ", mode:" << mode << ", dest:"<<sdest;
889  }
890  streamVal[mode]=sDestsValue;
891  }
892  (*transferSystemJson_)[psKeyItr->first] = streamVal;
893  }
894  }
895  }
896  else {
897  if (requireTSPSet_)
898  throw cms::Exception("EvFDaqDirector") << "transferSystem PSet not found";
899  }
900  }
901 
903  {
904  std::string streamRequestName;
905  if (transferSystemJson_->isMember(stream.c_str()))
906  streamRequestName = stream;
907  else {
908  std::stringstream msg;
909  msg << "Transfer system mode definitions missing for -: " << stream;
910  if (requireTSPSet_)
911  throw cms::Exception("EvFDaqDirector") << msg.str();
912  else {
913  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
914  return std::string("Failsafe");
915  }
916  }
917  //return empty if strict check parameter is not on
919  edm::LogWarning("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter."
920  << "Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
921  return std::string("Failsafe");
922  }
924  throw cms::Exception("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter.";
925  }
926  //check if stream has properly listed transfer stream
927  if (!transferSystemJson_->get(streamRequestName, "").isMember(selectedTransferMode_.c_str()))
928  {
929  std::stringstream msg;
930  msg << "Selected transfer mode " << selectedTransferMode_ << " is not specified for stream " << streamRequestName;
931  if (requireTSPSet_)
932  throw cms::Exception("EvFDaqDirector") << msg.str();
933  else
934  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
935  return std::string("Failsafe");
936  }
937  Json::Value destsVec = transferSystemJson_->get(streamRequestName, "").get(selectedTransferMode_,"");
938 
939  //flatten string json::Array into CSV std::string
940  std::string ret;
941  for (Json::Value::iterator it = destsVec.begin(); it!=destsVec.end(); it++)
942  {
943  if (ret!="") ret +=",";
944  ret+=(*it).asString();
945  }
946  return ret;
947  }
948 
950  {
951  if (mergeTypePset_.empty()) return;
952  if(!mergeTypeMap_.empty()) return;
954  if (topPset.existsAs<edm::ParameterSet>(mergeTypePset_,true))
955  {
956  const edm::ParameterSet& tsPset(topPset.getParameterSet(mergeTypePset_));
957  for (std::string pname : tsPset.getParameterNames()) {
958  std::string streamType = tsPset.getParameter<std::string>(pname);
959  tbb::concurrent_hash_map<std::string,std::string>::accessor ac;
960  mergeTypeMap_.insert(ac,pname);
961  ac->second = streamType;
962  ac.release();
963  }
964  }
965  }
966 
968  {
969  tbb::concurrent_hash_map<std::string,std::string>::const_accessor search_ac;
970  if (mergeTypeMap_.find(search_ac,stream))
971  return search_ac->second;
972 
973  edm::LogInfo("EvFDaqDirector") << " No merging type specified for stream " << stream << ". Using default value";
974  std::string defaultName = MergeTypeNames_[defaultType];
975  tbb::concurrent_hash_map<std::string,std::string>::accessor ac;
976  mergeTypeMap_.insert(ac,stream);
977  ac->second = defaultName;
978  ac.release();
979  return defaultName;
980  }
981 
983  std::string proc_flag = run_dir_ + "/processing";
984  int proc_flag_fd = open(proc_flag.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
985  close(proc_flag_fd);
986  }
987 
988  struct flock EvFDaqDirector::make_flock(short type, short whence, off_t start, off_t len, pid_t pid)
989  {
990 #ifdef __APPLE__
991  return {start, len, pid, type, whence};
992 #else
993  return {type, whence, start, len, pid};
994 #endif
995  }
996 
997 }
998 
#define LogDebug(id)
unsigned int nThreads_
type
Definition: HCALResponse.h:21
unsigned int maxNumberOfThreads() const
Definition: SystemBounds.h:46
T getParameter(std::string const &) const
std::string getStreamDestinations(std::string const &stream) const
Value get(UInt index, const Value &defaultValue) const
std::vector< std::string > & getData()
Definition: DataPoint.h:58
struct flock fu_rw_flk
static struct flock make_flock(short type, short whence, off_t start, off_t len, pid_t pid)
std::string run_string_
std::string protocolBufferHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string bolsFileName(const unsigned int run, const unsigned int ls)
std::string streamerDataChecksumFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string getMergedProtocolBufferHistogramFilePath(const unsigned int ls, std::string const &stream) const
void watchPreallocate(Preallocate::slot_type const &iSlot)
const_iterator begin() 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::list< std::pair< int, InputFile * > > * filesToDeletePtr_
static void fillDescriptions(edm::ConfigurationDescriptions &descriptions)
ParameterDescriptionBase * addUntracked(U const &iLabel, T const &value)
def create(alignables, pedeDump, additionalData, outputFile, config)
std::string getMergedDatChecksumFilePath(const unsigned int ls, std::string const &stream) const
std::shared_ptr< Json::Value > transferSystemJson_
void setAllowAnything()
allow any parameter label/value pairs
void openFULockfileStream(std::string &fuLockFilePath, bool create)
void accumulateFileSize(unsigned int lumi, unsigned long fileSize)
void watchPreGlobalEndLumi(PreGlobalEndLumi::slot_type const &iSlot)
std::string getInitFilePath(std::string const &stream) const
std::string getMergedRootHistogramFilePath(const unsigned int ls, std::string const &stream) const
std::string inputRawFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
ParameterSet const & getParameterSet(ParameterSetID const &id)
pthread_mutex_t init_lock_
LuminosityBlockID const & luminosityBlockID() const
Definition: GlobalContext.h:52
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 getOpenInitFilePath(std::string const &stream) const
std::string getOpenRawFilePath(const unsigned int ls, const unsigned int index) const
Value & append(const Value &value)
Append value to array at the end.
void createProcessingNotificationMaybe() 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
std::string getEoLSFilePathOnBU(const unsigned int ls) const
#define nullptr
std::string getEoRFilePath() const
unsigned long previousFileSize_
ParameterSetID const & parameterSetID() const
struct flock fu_rw_fulk
unsigned int maxNumberOfStreams() const
Definition: SystemBounds.h:43
std::string hltSourceDirectory_
std::string getOpenProtocolBufferHistogramFilePath(const unsigned int ls, std::string const &stream) const
void setComment(std::string const &value)
std::mutex * fileDeleteLockPtr_
tbb::concurrent_hash_map< std::string, std::string > mergeTypeMap_
std::string mergeTypePset_
std::string streamerDataFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
def chmod(path, mode)
Definition: eostools.py:293
std::string getMergedDatFilePath(const unsigned int ls, std::string const &stream) const
std::string getStreamMergeType(std::string const &stream, MergeType defaultType)
FileStatus updateFuLock(unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint64_t &lockWaitTime)
std::string getOpenOutputJsonFilePath(const unsigned int ls, std::string const &stream) const
std::string getOpenRootHistogramFilePath(const unsigned int ls, std::string const &stream) const
std::string getRootHistogramFilePath(const unsigned int ls, std::string const &stream) const
std::string getOpenDatFilePath(const unsigned int ls, std::string const &stream) const
std::string inputJsonFileName(const unsigned int run, const unsigned int ls, const unsigned int index)
std::string stopFilePath_
std::string bu_base_dir_
void removeFile(unsigned int ls, unsigned int index)
unsigned int stop_ls_override_
std::string selectedTransferMode_
std::string streamerJsonFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string getOpenInputJsonFilePath(const unsigned int ls, const unsigned int index) const
std::string eorFileName(const unsigned int run)
int readLastLSEntry(std::string const &file)
std::string stopFilePathPid_
unsigned int eolsNFilesIndex_
std::string getDatFilePath(const unsigned int ls, std::string const &stream) const
void preBeginJob(edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &)
void watchPreGlobalBeginRun(PreGlobalBeginRun::slot_type const &iSlot)
std::string getOutputJsonFilePath(const unsigned int ls, std::string const &stream) const
int getNFilesFromEoLS(std::string BUEoLSFile)
def ls(path, rec=False)
Definition: eostools.py:348
void preBeginRun(edm::GlobalContext const &globalContext)
std::string rootHistogramFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
Int asInt() const
std::string initFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
unsigned long long uint64_t
Definition: Time.h:15
void checkMergeTypePSet(edm::ProcessContext const &pc)
ParameterSet const & getParameterSet(std::string const &) const
void preGlobalEndLumi(edm::GlobalContext const &globalContext)
auto dp
Definition: deltaR.h:22
void deserialize(Json::Value &root) override
Definition: DataPoint.cc:56
void watchPostGlobalEndRun(PostGlobalEndRun::slot_type const &iSlot)
LuminosityBlockNumber_t luminosityBlock() const
void add(std::string const &label, ParameterSetDescription const &psetDescription)
std::string & getDefinition()
Definition: DataPoint.h:59
def remove(d, key, TELL=False)
Definition: MatrixUtil.py:210
tuple msg
Definition: mps_check.py:277
evf::FastMonitoringService * fms_
void watchPreBeginJob(PreBeginJob::slot_type const &iSlot)
convenience function for attaching to signal
std::string bu_run_dir_
std::string rootHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
def mkdir(path)
Definition: eostools.py:250
std::string findHighestRunDir()
Definition: DirManager.cc:20
std::string getBoLSFilePathOnFU(const unsigned int ls) const
char data[epos_bytes_allocation]
Definition: EPOS_Wrapper.h:82
void postEndRun(edm::GlobalContext const &globalContext)
Unserialize a JSON document into a Value.
Definition: reader.h:16
static const std::vector< std::string > MergeTypeNames_
const_iterator end() const
void checkTransferSystemPSet(edm::ProcessContext const &pc)
std::vector< std::string > const & getNames()
std::string getEoLSFilePathOnFU(const unsigned int ls) const
bool bumpFile(unsigned int &ls, unsigned int &index, std::string &nextFile, uint32_t &fsize, int maxLS)
void preallocate(edm::service::SystemBounds const &bounds)
Iterator for object and array value.
Definition: value.h:1007
std::string getInputJsonFilePath(const unsigned int ls, const unsigned int index) const
def check(config)
Definition: trackerTree.py:14
std::string getRunOpenDirPath() const
JetCorrectorParameters::Definitions def
Definition: classes.h:6
std::string protocolBufferHistogramFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
unsigned int fuLockPollInterval_
std::string eolsFileName(const unsigned int run, const unsigned int ls)
unsigned int nStreams_
std::string streamerDataFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
std::string getEoRFilePathOnFU() const
std::string bu_run_open_dir_
array value (ordered list)
Definition: value.h:31