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.
8 
14 
15 #include <iostream>
16 #include <sstream>
17 #include <sys/time.h>
18 #include <unistd.h>
19 #include <stdio.h>
20 #include <sys/file.h>
21 #include <boost/lexical_cast.hpp>
22 #include <boost/filesystem/fstream.hpp>
23 
24 //#define DEBUG
25 
26 namespace evf {
27 
28  namespace {
29  struct flock make_flock(short type, short whence, off_t start, off_t len, pid_t pid)
30  {
31 #ifdef __APPLE__
32  return {start, len, pid, type, whence};
33 #else
34  return {type, whence, start, len, pid};
35 #endif
36  }
37  }
38 
40  edm::ActivityRegistry& reg) :
41  testModeNoBuilderUnit_(
42  pset.getUntrackedParameter<bool> ("testModeNoBuilderUnit",
43  false)
44  ),
45  base_dir_(
46  pset.getUntrackedParameter<std::string> ("baseDir", "/data")
47  ),
48  bu_base_dir_(
49  pset.getUntrackedParameter<std::string> ("buBaseDir", "/data")
50  ),
51  directorBu_(
52  pset.getUntrackedParameter<bool> ("directorIsBu", false)
53  ),
54  run_(pset.getUntrackedParameter<unsigned int> ("runNumber",0)),
55  outputAdler32Recheck_(pset.getUntrackedParameter<bool>("outputAdler32Recheck",false)),
56  requireTSPSet_(pset.getUntrackedParameter<bool>("requireTransfersPSet",false)),
57  selectedTransferMode_(pset.getUntrackedParameter<std::string>("selectedTransferMode","")),
58  hltSourceDirectory_(pset.getUntrackedParameter<std::string>("hltSourceDirectory","")),
59  fuLockPollInterval_(pset.getUntrackedParameter<unsigned int>("fuLockPollInterval",2000)),
60  hostname_(""),
61  bu_readlock_fd_(-1),
62  bu_writelock_fd_(-1),
63  fu_readwritelock_fd_(-1),
64  data_readwrite_fd_(-1),
65  fulocal_rwlock_fd_(-1),
66  fulocal_rwlock_fd2_(-1),
67 
68  bu_w_lock_stream(0),
69  bu_r_lock_stream(0),
70  fu_rw_lock_stream(0),
71  //bu_w_monitor_stream(0),
72  //bu_t_monitor_stream(0),
73  data_rw_stream(0),
74 
75  dirManager_(base_dir_),
76 
77  previousFileSize_(0),
78  jumpLS_(0),
79  jumpIndex_(0),
80 
81  bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
82  bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
83  bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
84  bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
85  fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
86  fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
87  data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
88  data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
89  //fulocal_rw_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
90  //fulocal_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
91  //fulocal_rw_flk2( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
92  //fulocal_rw_fulk2( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
93  {
94 
100 
101  std::stringstream ss;
102  ss << "run" << std::setfill('0') << std::setw(6) << run_;
103  run_string_ = ss.str();
105 
106  //save hostname for later
107  char hostname[33];
108  gethostname(hostname,32);
109  hostname_ = hostname;
110 
111  char * fuLockPollIntervalPtr = getenv("FFF_LOCKPOLLINTERVAL");
112  if (fuLockPollIntervalPtr) {
113  try {
114  fuLockPollInterval_=boost::lexical_cast<unsigned int>(std::string(fuLockPollIntervalPtr));
115  edm::LogInfo("Setting fu lock poll interval by environment string: ") << fuLockPollInterval_ << " us";
116  }
117  catch (...) {edm::LogWarning("Unable to parse environment string: ") << std::string(fuLockPollIntervalPtr);}
118  }
119 
120  // check if base dir exists or create it accordingly
121  int retval = mkdir(base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
122  if (retval != 0 && errno != EEXIST) {
123  throw cms::Exception("DaqDirector") << " Error checking for base dir -: "
124  << base_dir_ << " mkdir error:" << strerror(errno);
125  }
126 
127  //create run dir in base dir
128  umask(0);
129  retval = mkdir(run_dir_.c_str(),
130  S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
131  if (retval != 0 && errno != EEXIST) {
132  throw cms::Exception("DaqDirector") << " Error creating run dir -: "
133  << run_dir_ << " mkdir error:" << strerror(errno);
134  }
135 
136  //create fu-local.lock in run open dir
137  if (!directorBu_) {
138 
140  std::string fulocal_lock_ = getRunOpenDirPath() +"/fu-local.lock";
141  fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
142  if (fulocal_rwlock_fd_==-1)
143  throw cms::Exception("DaqDirector") << " Error creating/opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
144  chmod(fulocal_lock_.c_str(),0777);
145  fsync(fulocal_rwlock_fd_);
146  //open second fd for another input source thread
147  fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
148  if (fulocal_rwlock_fd2_==-1)
149  throw cms::Exception("DaqDirector") << " Error opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
150  }
151 
152  //bu_run_dir: for FU, for which the base dir is local and the BU is remote, it is expected to be there
153  //for BU, it is created at this point
154  if (directorBu_)
155  {
157  std::string bulockfile = bu_run_dir_ + "/bu.lock";
158  std::string fulockfile = bu_run_dir_ + "/fu.lock";
159 
160  //make or find bu run dir
161  retval = mkdir(bu_run_dir_.c_str(),
162  S_IRWXU | S_IRWXG | S_IRWXO);
163  if (retval != 0 && errno != EEXIST) {
164  throw cms::Exception("DaqDirector")
165  << " Error creating bu run dir -: " << bu_run_dir_
166  << " mkdir error:" << strerror(errno) << "\n";
167  }
168  bu_run_open_dir_ = bu_run_dir_ + "/open";
169  retval = mkdir(bu_run_open_dir_.c_str(),
170  S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
171  if (retval != 0 && errno != EEXIST) {
172  throw cms::Exception("DaqDirector") << " Error creating bu run open dir -: "
173  << bu_run_open_dir_ << " mkdir error:" << strerror(errno)
174  << "\n";
175  }
176 
177  // the BU director does not need to know about the fu lock
178  bu_writelock_fd_ = open(bulockfile.c_str(),
179  O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
180  if (bu_writelock_fd_ == -1)
181  edm::LogWarning("EvFDaqDirector") << "problem with creating filedesc for buwritelock -: "
182  << strerror(errno);
183  else
184  edm::LogInfo("EvFDaqDirector") << "creating filedesc for buwritelock -: "
185  << bu_writelock_fd_;
186  bu_w_lock_stream = fdopen(bu_writelock_fd_, "w");
187  if (bu_w_lock_stream == 0)
188  edm::LogWarning("EvFDaqDirector")<< "Error creating write lock stream -: " << strerror(errno);
189 
190  // BU INITIALIZES LOCK FILE
191  // FU LOCK FILE OPEN
192  openFULockfileStream(fulockfile, true);
194  fflush(fu_rw_lock_stream);
195  close(fu_readwritelock_fd_);
196 
197  if (hltSourceDirectory_.size())
198  {
199  struct stat buf;
200  if (stat(hltSourceDirectory_.c_str(),&buf)==0) {
201  std::string hltdir=bu_run_dir_+"/hlt";
202  std::string tmphltdir=bu_run_open_dir_+"/hlt";
203  retval = mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
204 
205  boost::filesystem::copy_file(hltSourceDirectory_+"/HltConfig.py",tmphltdir+"/HltConfig.py");
206  try {
207  boost::filesystem::copy_file(hltSourceDirectory_+"/CMSSW_VERSION",tmphltdir+"/CMSSW_VERSION");
208  boost::filesystem::copy_file(hltSourceDirectory_+"/SCRAM_ARCH",tmphltdir+"/SCRAM_ARCH");
209  } catch (...) {}
210 
211  boost::filesystem::copy_file(hltSourceDirectory_+"/fffParameters.jsn",tmphltdir+"/fffParameters.jsn");
212 
213  boost::filesystem::rename(tmphltdir,hltdir);
214  }
215  else
216  throw cms::Exception("DaqDirector") << " Error looking for HLT configuration -: " << hltSourceDirectory_;
217  }
218  //else{}//no configuration specified
219  }
220  else
221  {
222  // for FU, check if bu base dir exists
223 
224  retval = mkdir(bu_base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
225  if (retval != 0 && errno != EEXIST) {
226  throw cms::Exception("DaqDirector") << " Error checking for bu base dir -: "
227  << bu_base_dir_ << " mkdir error:" << strerror(errno) << "\n";
228  }
229 
231  std::string fulockfile = bu_run_dir_ + "/fu.lock";
232  openFULockfileStream(fulockfile, false);
233  }
234 
235  pthread_mutex_init(&init_lock_,NULL);
236 
237  stopFilePath_ = run_dir_+"/CMSSW_STOP";
238  }
239 
241  {
242  if (fulocal_rwlock_fd_!=-1) {
243  unlockFULocal();
244  close(fulocal_rwlock_fd_);
245  }
246 
247  if (fulocal_rwlock_fd2_!=-1) {
248  unlockFULocal2();
249  close(fulocal_rwlock_fd2_);
250  }
251 
252  }
253 
254  void EvFDaqDirector::postEndRun(edm::GlobalContext const& globalContext) {
255  close(bu_readlock_fd_);
256  close(bu_writelock_fd_);
257  if (directorBu_) {
258  std::string filename = bu_run_dir_ + "/bu.lock";
259  removeFile(filename);
260  }
261  }
262 
264 
265  for (unsigned int i=0;i<bounds.maxNumberOfStreams();i++){
266  streamFileTracker_.push_back(-1);
267  }
268  nThreads_=bounds.maxNumberOfStreams();
269  nStreams_=bounds.maxNumberOfThreads();
270 
272  }
273 
274  void EvFDaqDirector::preBeginRun(edm::GlobalContext const& globalContext) {
275 
276  //assert(run_ == id.run());
277 
278  // check if the requested run is the latest one - issue a warning if it isn't
280  edm::LogWarning("EvFDaqDirector") << "WARNING - checking run dir -: "
281  << run_dir_ << ". This is not the highest run "
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 (...) {/*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 (...) {/*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  inline void EvFDaqDirector::preSourceEvent(edm::StreamID const& streamID) {
332  }
333 
334 
335  std::string EvFDaqDirector::getInputJsonFilePath(const unsigned int ls, const unsigned int index) const {
336  return bu_run_dir_ + "/" + fffnaming::inputJsonFileName(run_,ls,index);
337  }
338 
339 
340  std::string EvFDaqDirector::getRawFilePath(const unsigned int ls, const unsigned int index) const {
341  return bu_run_dir_ + "/" + fffnaming::inputRawFileName(run_,ls,index);
342  }
343 
344  std::string EvFDaqDirector::getOpenRawFilePath(const unsigned int ls, const unsigned int index) const {
345  return bu_run_dir_ + "/open/" + fffnaming::inputRawFileName(run_,ls,index);
346  }
347 
348  std::string EvFDaqDirector::getOpenInputJsonFilePath(const unsigned int ls, const unsigned int index) const {
349  return bu_run_dir_ + "/open/" + fffnaming::inputJsonFileName(run_,ls,index);
350  }
351 
352  std::string EvFDaqDirector::getOpenDatFilePath(const unsigned int ls, std::string const& stream) const {
353  return run_dir_ + "/open/" + fffnaming::streamerDataFileNameWithPid(run_,ls,stream);
354  }
355 
356  std::string EvFDaqDirector::getOpenOutputJsonFilePath(const unsigned int ls, std::string const& stream) const {
357  return run_dir_ + "/open/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
358  }
359 
360  std::string EvFDaqDirector::getOutputJsonFilePath(const unsigned int ls, std::string const& stream) const {
361  return run_dir_ + "/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
362  }
363 
364  std::string EvFDaqDirector::getMergedDatFilePath(const unsigned int ls, std::string const& stream) const {
366  }
367 
368  std::string EvFDaqDirector::getMergedDatChecksumFilePath(const unsigned int ls, std::string const& stream) const {
370  }
371 
373  return run_dir_ + "/open/" + fffnaming::initFileNameWithPid(run_,0,stream);
374  }
375 
377  return run_dir_ + "/" + fffnaming::initFileNameWithPid(run_,0,stream);
378  }
379 
381  return run_dir_ + "/open/" + fffnaming::protocolBufferHistogramFileNameWithPid(run_,ls,stream);
382  }
383 
386  }
387 
390  }
391 
392  std::string EvFDaqDirector::getOpenRootHistogramFilePath(const unsigned int ls, std::string const& stream) const {
393  return run_dir_ + "/open/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
394  }
395 
396  std::string EvFDaqDirector::getRootHistogramFilePath(const unsigned int ls, std::string const& stream) const {
397  return run_dir_ + "/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
398  }
399 
402  }
403 
405  return bu_run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
406  }
407 
409  return run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
410  }
411 
413  return run_dir_ + "/" + fffnaming::bolsFileName(run_,ls);
414  }
415 
417  return bu_run_dir_ + "/" + fffnaming::eorFileName(run_);
418  }
419 
420 
422  return run_dir_ + "/" + fffnaming::eorFileName(run_);
423  }
424 
426  int retval = remove(filename.c_str());
427  if (retval != 0)
428  edm::LogError("EvFDaqDirector") << "Could not remove used file -: " << filename << ". error = "
429  << strerror(errno);
430  }
431 
432  void EvFDaqDirector::removeFile(unsigned int ls, unsigned int index) {
433  removeFile(getRawFilePath(ls,index));
434  }
435 
436  EvFDaqDirector::FileStatus EvFDaqDirector::updateFuLock(unsigned int& ls, std::string& nextFile, uint32_t& fsize, uint64_t& lockWaitTime) {
437  EvFDaqDirector::FileStatus fileStatus = noFile;
438 
439  int retval = -1;
440  int lock_attempts = 0;
441 
442  struct stat buf;
443  int stopFileLS = -1;
444  if (stat(stopFilePath_.c_str(),&buf)==0) {
445  stopFileLS = readLastLSEntry(stopFilePath_);
446  edm::LogWarning("EvFDaqDirector") << "Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
447  //return runEnded;
448  }
449 
450  timeval ts_lockbegin;
451  gettimeofday(&ts_lockbegin,0);
452 
453  while (retval==-1) {
454  retval = fcntl(fu_readwritelock_fd_, F_SETLK, &fu_rw_flk);
455  if (retval==-1) usleep(fuLockPollInterval_);
456  else continue;
457 
458  lock_attempts+=fuLockPollInterval_;
459  if (lock_attempts>5000000 || errno==116) {
460  if (errno==116)
461  edm::LogWarning("EvFDaqDirector") << "Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
462  else
463  edm::LogWarning("EvFDaqDirector") << "Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno "
464  << errno <<":"<< strerror(errno) << std::endl;
465 
466 
467  if (stat(getEoLSFilePathOnFU(ls).c_str(),&buf)==0) {
468  edm::LogWarning("EvFDaqDirector") << "Detected local EoLS for lumisection "<< ls ;
469  ls++;
470  return noFile;
471  }
472 
473  if (stat(bu_run_dir_.c_str(), &buf)!=0) return runEnded;
474  if (stat((bu_run_dir_+"/fu.lock").c_str(), &buf)!=0) return runEnded;
475  lock_attempts=0;
476  }
477  }
478 
479  timeval ts_lockend;
480  gettimeofday(&ts_lockend,0);
481  long deltat = (ts_lockend.tv_usec-ts_lockbegin.tv_usec) + (ts_lockend.tv_sec-ts_lockbegin.tv_sec)*1000000;
482  if (deltat>0.) lockWaitTime=deltat;
483 
484 
485 
486  if(retval!=0) return fileStatus;
487 
488 #ifdef DEBUG
489  timeval ts_lockend;
490  gettimeofday(&ts_lockend,0);
491 #endif
492 
493  // if the stream is readable
494  if (fu_rw_lock_stream != 0) {
495  unsigned int readLs, readIndex, jumpLs, jumpIndex;
496  int check = 0;
497  // rewind the stream
498  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
499  // if rewinded ok
500  if (check == 0) {
501  // read its' values
503  fscanf(fu_rw_lock_stream, "%u %u %u %u", &readLs, &readIndex,
504  &jumpLs, &jumpIndex);
505  else {
506  fscanf(fu_rw_lock_stream, "%u %u", &readLs, &readIndex);
507  edm::LogInfo("EvFDaqDirector") << "Read fu.lock file file -: " << readLs << ":" << readIndex;
508  }
509 
510  // try to bump
511  bool bumpedOk = bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
512  ls = readLs;
513  // there is a new file to grab or lumisection ended
514  if (bumpedOk) {
515  // write new data
516  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
517  if (check == 0) {
518  ftruncate(fu_readwritelock_fd_, 0);
519  // write next index in the file, which is the file the next process should take
521  fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs,
522  readIndex + 1, readLs + 2, readIndex + 1);
523  jumpLS_ = readLs + 2;
525  } else {
526  fprintf(fu_rw_lock_stream, "%u %u", readLs,
527  readIndex + 1);
528  }
529  fflush(fu_rw_lock_stream);
530  fsync(fu_readwritelock_fd_);
531  fileStatus = newFile;
532 
534  edm::LogInfo("EvFDaqDirector") << "Written to file -: " << readLs << ":"
535  << readIndex + 1 << " --> " << readLs + 2
536  << ":" << readIndex + 1;
537  else
538  LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":"
539  << readIndex + 1;
540 
541  } else
542  throw cms::Exception("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
543  << strerror(errno);
544  }
545  } else
546  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for reading failed with error "
547  << strerror(errno);
548  } else
549  edm::LogError("EvFDaqDirector") << "fu read/write lock stream is invalid " << strerror(errno);
550 
551 #ifdef DEBUG
552  timeval ts_preunlock;
553  gettimeofday(&ts_preunlock,0);
554  int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
555  double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
556 #endif
557 
558  //if new json is present, lock file which FedRawDataInputSource will later unlock
559  if (fileStatus==newFile) lockFULocal();
560 
561  //release lock at this point
562  int retvalu=-1;
563  retvalu=fcntl(fu_readwritelock_fd_, F_SETLKW, &fu_rw_fulk);
564  if (retvalu==-1) edm::LogError("EvFDaqDirector") << "Error unlocking the fu.lock " << strerror(errno);
565 
566 #ifdef DEBUG
567  edm::LogDebug("EvFDaqDirector") << "Waited during lock -: " << locked_period << " seconds";
568 #endif
569 
570  if ( fileStatus == noFile ) {
571  struct stat buf;
572  //edm::LogInfo("EvFDaqDirector") << " looking for EoR file: " << getEoRFilePath().c_str();
573  if ( stat(getEoRFilePath().c_str(), &buf) == 0 || stat(bu_run_dir_.c_str(), &buf)!=0)
574  fileStatus = runEnded;
575  if (stopFileLS>=0 && (int)ls > stopFileLS) {
576  edm::LogInfo("EvFDaqDirector") << "Reached maximum lumisection set by hltd";
577  fileStatus = runEnded;
578  }
579  }
580  return fileStatus;
581  }
582 
584 
585  boost::filesystem::ifstream ij(BUEoLSFile);
586  Json::Value deserializeRoot;
588 
589  if (!reader.parse(ij, deserializeRoot)) {
590  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << BUEoLSFile;
591  return -1;
592  }
593 
595  DataPoint dp;
596  dp.deserialize(deserializeRoot);
597 
598  //read definition
599  if (readEolsDefinition_) {
600  //std::string def = boost::algorithm::trim(dp.getDefinition());
602  if (!def.size()) readEolsDefinition_=false;
603  while (def.size()) {
604  std::string fullpath;
605  if (def.find('/')==0)
606  fullpath = def;
607  else
608  fullpath = bu_run_dir_+'/'+def;
609  struct stat buf;
610  if (stat(fullpath.c_str(), &buf) == 0) {
611  DataPointDefinition eolsDpd;
612  std::string defLabel = "legend";
613  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
614  if (eolsDpd.getNames().size()==0) {
615  //try with "data" label if "legend" format is not used
616  eolsDpd = DataPointDefinition();
617  defLabel="data";
618  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
619  }
620  for (unsigned int i=0;i<eolsDpd.getNames().size();i++)
621  if (eolsDpd.getNames().at(i)=="NFiles")
623  readEolsDefinition_=false;
624  break;
625  }
626  //check if we can still find definition
627  if (def.size()<=1 || def.find('/')==std::string::npos) {
628  readEolsDefinition_=false;
629  break;
630  }
631  def = def.substr(def.find('/')+1);
632  }
633  }
634 
635  if (dp.getData().size()>eolsNFilesIndex_)
636  data = dp.getData()[eolsNFilesIndex_];
637  else {
638  edm::LogError("EvFDaqDirector") << " error reading number of files from BU JSON -: " << BUEoLSFile;
639  return -1;
640  }
641  return boost::lexical_cast<int>(data);
642  }
643 
644  bool EvFDaqDirector::bumpFile(unsigned int& ls, unsigned int& index, std::string& nextFile, uint32_t& fsize, int maxLS) {
645 
646  if (previousFileSize_ != 0) {
647  if (!fms_) {
648  try {
650  } catch (...) {
651  edm::LogError("EvFDaqDirector") <<" FastMonitoringService not found";
652  }
653  }
655  previousFileSize_ = 0;
656  }
657 
658  //reached limit
659  if (maxLS>=0 && ls > (unsigned int)maxLS) return false;
660 
661  struct stat buf;
662  std::stringstream ss;
663  unsigned int nextIndex = index;
664  nextIndex++;
665 
666  // 1. Check suggested file
667  nextFile = getInputJsonFilePath(ls,index);
668  if (stat(nextFile.c_str(), &buf) == 0) {
669 
670  previousFileSize_ = buf.st_size;
671  fsize = buf.st_size;
672  return true;
673  }
674  // 2. No file -> lumi ended? (and how many?)
675  else {
676  std::string BUEoLSFile = getEoLSFilePathOnBU(ls);
677  bool eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
678  unsigned int startingLumi = ls;
679  while (eolFound) {
680 
681  // recheck that no raw file appeared in the meantime
682  if (stat(nextFile.c_str(), &buf) == 0) {
683  previousFileSize_ = buf.st_size;
684  fsize = buf.st_size;
685  return true;
686  }
687 
688  int indexFilesInLS = getNFilesFromEoLS(BUEoLSFile);
689  if (indexFilesInLS < 0)
690  //parsing failed
691  return false;
692  else {
693  //check index
694  if ((int)index<indexFilesInLS) {
695  //we have 2 files, and check for 1 failed... retry (2 will never be here)
696  edm::LogError("EvFDaqDirector") << "Potential miss of index file in LS -: " << ls << ". Missing "
697  << nextFile << " because " << indexFilesInLS-1 << " is the highest index expected. Will not update fu.lock file";
698  return false;
699  }
700  }
701  // this lumi ended, check for files
702  ++ls;
703  index = 0;
704 
705  //reached limit
706  if (maxLS>=0 && ls > (unsigned int)maxLS) return false;
707 
708  nextFile = getInputJsonFilePath(ls,0);
709  if (stat(nextFile.c_str(), &buf) == 0) {
710  // a new file was found at new lumisection, index 0
711  previousFileSize_ = buf.st_size;
712  fsize = buf.st_size;
713 
715  // rename ended lumi to + 2
716  std::string sourceEol = getEoLSFilePathOnBU(startingLumi);
717 
718  std::string destEol = getEoLSFilePathOnBU(startingLumi+2);
719 
720  std::string cpCmd = "cp " + sourceEol + " " + destEol;
721  edm::LogInfo("EvFDaqDirector") << " testmode: Running copy cmd -: " << cpCmd;
722  int rc = system(cpCmd.c_str());
723  if (rc != 0) {
724  edm::LogError("EvFDaqDirector") << " testmode: COPY EOL FAILED!!!!! -: " << cpCmd;
725  }
726  }
727 
728  return true;
729  }
730  BUEoLSFile = getEoLSFilePathOnBU(ls);
731  eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
732  }
733  }
734  // no new file found
735  return false;
736  }
737 
739  if (fu_rw_lock_stream == 0)
740  edm::LogError("EvFDaqDirector") << "Error creating fu read/write lock stream "
741  << strerror(errno);
742  else {
743  edm::LogInfo("EvFDaqDirector") << "Initializing FU LOCK FILE";
744  unsigned int readLs = 1, readIndex = 0, jumpLs = 3, jumpIndex = 0;
746  fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs, readIndex,
747  jumpLs, jumpIndex);
748  else
749  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
750  }
751  }
752 
754  if (create) {
755  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
756  S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
757  chmod(fulockfile.c_str(),0766);
758  } else {
759  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
760  }
761  if (fu_readwritelock_fd_ == -1)
762  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
763  << " create:" << create << " error:" << strerror(errno);
764  else
765  LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: "
767 
768  fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
769  }
770 
771  //create if does not exist then lock the merge destination file
773  data_rw_stream = fopen(getMergedDatFilePath(ls,stream).c_str(), "a"); //open stream for appending
775  if (data_readwrite_fd_ == -1)
776  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for datamerge "
777  << strerror(errno);
778  else
779  LogDebug("EvFDaqDirector") << "creating filedesc for datamerge -: "
781  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_flk);
782 
783  return data_rw_stream;
784  }
785 
787  fflush(data_rw_stream);
788  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_fulk);
789  fclose(data_rw_stream);
790  }
791 
793  pthread_mutex_lock(&init_lock_);
794  }
795 
797  pthread_mutex_unlock(&init_lock_);
798  }
799 
801  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_flk);
802  flock(fulocal_rwlock_fd_,LOCK_SH);
803  }
804 
806  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_fulk);
807  flock(fulocal_rwlock_fd_,LOCK_UN);
808  }
809 
810 
812  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_flk2);
813  flock(fulocal_rwlock_fd2_,LOCK_EX);
814  }
815 
817  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_fulk2);
818  flock(fulocal_rwlock_fd2_,LOCK_UN);
819  }
820 
821 
823  // create open dir if not already there
824 
826  if (!boost::filesystem::is_directory(openPath)) {
827  LogDebug("EvFDaqDirector") << "<open> FU dir not found. Creating... -:" << openPath.string();
828  boost::filesystem::create_directories(openPath);
829  }
830  }
831 
832 
834 
835  boost::filesystem::ifstream ij(file);
836  Json::Value deserializeRoot;
838 
839  if (!reader.parse(ij, deserializeRoot)) {
840  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << file;
841  return -1;
842  }
843 
844  int ret = deserializeRoot.get("lastLS","").asInt();
845  return ret;
846 
847  }
848 
849  //if transferSystem PSet is present in the menu, we require it to be complete and consistent for all specified streams
851  {
852  transferSystemJson_.reset(new Json::Value);
853  if (edm::getProcessParameterSet().existsAs<edm::ParameterSet>("transferSystem",true))
854  {
855  const edm::ParameterSet& tsPset(edm::getProcessParameterSet().getParameterSet("transferSystem"));
856 
857  Json::Value destinationsVal(Json::arrayValue);
858  std::vector<std::string> destinations = tsPset.getParameter<std::vector<std::string>>("destinations");
859  for (auto & dest: destinations) destinationsVal.append(dest);
860  (*transferSystemJson_)["destinations"]=destinationsVal;
861 
862  Json::Value modesVal(Json::arrayValue);
863  std::vector<std::string> modes = tsPset.getParameter< std::vector<std::string> >("transferModes");
864  for (auto & mode: modes) modesVal.append(mode);
865  (*transferSystemJson_)["transferModes"]=modesVal;
866 
867  for (auto psKeyItr =tsPset.psetTable().begin();psKeyItr!=tsPset.psetTable().end(); ++ psKeyItr) {
868  if (psKeyItr->first!="destinations" && psKeyItr->first!="transferModes") {
869  const edm::ParameterSet & streamDef = tsPset.getParameterSet(psKeyItr->first);
870  Json::Value streamVal;
871  for (auto & mode : modes) {
872  //validation
873  if (!streamDef.existsAs<std::vector<std::string>>(mode,true))
874  throw cms::Exception("EvFDaqDirector") << " Missing transfer system specification for -:" << psKeyItr->first << " (transferMode " << mode << ")";
875  std::vector<std::string> streamDestinations = streamDef.getParameter<std::vector<std::string>>(mode);
876 
877  Json::Value sDestsValue(Json::arrayValue);
878 
879  if (!streamDestinations.size())
880  throw cms::Exception("EvFDaqDirector") << " Missing transter system destination(s) for -: "<< psKeyItr->first << ", mode:" << mode;
881 
882  for (auto & sdest:streamDestinations) {
883  bool sDestValid=false;
884  sDestsValue.append(sdest);
885  for (auto & dest: destinations) {
886  if (dest==sdest) sDestValid=true;
887  }
888  if (!sDestValid)
889  throw cms::Exception("EvFDaqDirector") << " Invalid transter system destination specified for -: "<< psKeyItr->first << ", mode:" << mode << ", dest:"<<sdest;
890  }
891  streamVal[mode]=sDestsValue;
892  }
893  (*transferSystemJson_)[psKeyItr->first] = streamVal;
894  }
895  }
896  }
897  else {
898  if (requireTSPSet_)
899  throw cms::Exception("EvFDaqDirector") << "transferSystem PSet not found";
900  }
901  }
902 
904  {
905  std::string streamRequestName;
906  if (transferSystemJson_->isMember(stream.c_str()))
907  streamRequestName = stream;
908  else {
909  std::stringstream msg;
910  msg << "Transfer system mode definitions missing for -: " << stream;
911  if (requireTSPSet_)
912  throw cms::Exception("EvFDaqDirector") << msg.str();
913  else {
914  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
915  return std::string("Failsafe");
916  }
917  }
918  //return empty if strict check parameter is not on
920  edm::LogWarning("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter."
921  << "Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
922  return std::string("Failsafe");
923  }
925  throw cms::Exception("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter.";
926  }
927  //check if stream has properly listed transfer stream
928  if (!transferSystemJson_->get(streamRequestName, "").isMember(selectedTransferMode_.c_str()))
929  {
930  std::stringstream msg;
931  msg << "Selected transfer mode " << selectedTransferMode_ << " is not specified for stream " << streamRequestName;
932  if (requireTSPSet_)
933  throw cms::Exception("EvFDaqDirector") << msg.str();
934  else
935  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
936  return std::string("Failsafe");
937  }
938  Json::Value destsVec = transferSystemJson_->get(streamRequestName, "").get(selectedTransferMode_,"");
939 
940  //flatten string json::Array into CSV std::string
942  for (Json::Value::iterator it = destsVec.begin(); it!=destsVec.end(); it++)
943  {
944  if (ret!="") ret +=",";
945  ret+=(*it).asString();
946  }
947  return ret;
948  }
949 
950 }
#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
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_
std::string getMergedDatChecksumFilePath(const unsigned int ls, std::string const &stream) const
std::shared_ptr< Json::Value > transferSystemJson_
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.
#define NULL
Definition: scimark2.h:8
struct flock data_rw_flk
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_
unsigned int jumpIndex_
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
std::mutex * fileDeleteLockPtr_
std::string streamerDataFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
tuple path
else: Piece not in the list, fine.
EvFDaqDirector(const edm::ParameterSet &pset, edm::ActivityRegistry &reg)
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_
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)
psettable const & psetTable() const
Definition: ParameterSet.h:257
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 & getProcessParameterSet()
Definition: Registry.cc:85
ParameterSet const & getParameterSet(std::string const &) const
void preGlobalEndLumi(edm::GlobalContext const &globalContext)
tuple pid
Definition: sysUtil.py:22
void watchPostGlobalEndRun(PostGlobalEndRun::slot_type const &iSlot)
LuminosityBlockNumber_t luminosityBlock() const
std::string & getDefinition()
Definition: DataPoint.h:59
evf::FastMonitoringService * fms_
std::string bu_run_dir_
std::string rootHistogramFileNameWithInstance(const unsigned int run, const unsigned int ls, std::string const &stream, std::string const &instance)
std::string findHighestRunDir()
Definition: DirManager.cc:20
std::string getBoLSFilePathOnFU(const unsigned int ls) const
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:120
const_iterator end() const
std::vector< std::string > const & getNames()
unsigned int jumpLS_
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