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