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