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  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  jumpLS_(0),
78  jumpIndex_(0),
79 
80  bu_w_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, 0 )),
81  bu_r_flk( make_flock( F_RDLCK, SEEK_SET, 0, 0, 0 )),
82  bu_w_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
83  bu_r_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, 0 )),
84  fu_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
85  fu_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
86  data_rw_flk( make_flock ( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
87  data_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
88  //fulocal_rw_flk( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
89  //fulocal_rw_fulk( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() )),
90  //fulocal_rw_flk2( make_flock( F_WRLCK, SEEK_SET, 0, 0, getpid() )),
91  //fulocal_rw_fulk2( make_flock( F_UNLCK, SEEK_SET, 0, 0, getpid() ))
92  {
93 
99 
100  std::stringstream ss;
101  ss << "run" << std::setfill('0') << std::setw(6) << run_;
102  run_string_ = ss.str();
104 
105  //save hostname for later
106  char hostname[33];
107  gethostname(hostname,32);
108  hostname_ = hostname;
109  // check if base dir exists or create it accordingly
110  int retval = mkdir(base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
111  if (retval != 0 && errno != EEXIST) {
112  throw cms::Exception("DaqDirector") << " Error checking for base dir -: "
113  << base_dir_ << " mkdir error:" << strerror(errno);
114  }
115 
116  //create run dir in base dir
117  umask(0);
118  retval = mkdir(run_dir_.c_str(),
119  S_IRWXU | S_IRWXG | S_IROTH | S_IRWXO | S_IXOTH);
120  if (retval != 0 && errno != EEXIST) {
121  throw cms::Exception("DaqDirector") << " Error creating run dir -: "
122  << run_dir_ << " mkdir error:" << strerror(errno);
123  }
124 
125  //create fu-local.lock in run open dir
126  if (!directorBu_) {
127 
129  std::string fulocal_lock_ = getRunOpenDirPath() +"/fu-local.lock";
130  fulocal_rwlock_fd_ = open(fulocal_lock_.c_str(), O_RDWR | O_CREAT, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
131  if (fulocal_rwlock_fd_==-1)
132  throw cms::Exception("DaqDirector") << " Error creating/opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
133  chmod(fulocal_lock_.c_str(),0777);
134  fsync(fulocal_rwlock_fd_);
135  //open second fd for another input source thread
136  fulocal_rwlock_fd2_ = open(fulocal_lock_.c_str(), O_RDWR, S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);//O_RDWR?
137  if (fulocal_rwlock_fd2_==-1)
138  throw cms::Exception("DaqDirector") << " Error opening a local lock file -: " << fulocal_lock_.c_str() << " : " << strerror(errno);
139  }
140 
141  //bu_run_dir: for FU, for which the base dir is local and the BU is remote, it is expected to be there
142  //for BU, it is created at this point
143  if (directorBu_)
144  {
146  std::string bulockfile = bu_run_dir_ + "/bu.lock";
147  std::string fulockfile = bu_run_dir_ + "/fu.lock";
148 
149  //make or find bu run dir
150  retval = mkdir(bu_run_dir_.c_str(),
151  S_IRWXU | S_IRWXG | S_IRWXO);
152  if (retval != 0 && errno != EEXIST) {
153  throw cms::Exception("DaqDirector")
154  << " Error creating bu run dir -: " << bu_run_dir_
155  << " mkdir error:" << strerror(errno) << "\n";
156  }
157  bu_run_open_dir_ = bu_run_dir_ + "/open";
158  retval = mkdir(bu_run_open_dir_.c_str(),
159  S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
160  if (retval != 0 && errno != EEXIST) {
161  throw cms::Exception("DaqDirector") << " Error creating bu run open dir -: "
162  << bu_run_open_dir_ << " mkdir error:" << strerror(errno)
163  << "\n";
164  }
165 
166  // the BU director does not need to know about the fu lock
167  bu_writelock_fd_ = open(bulockfile.c_str(),
168  O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU);
169  if (bu_writelock_fd_ == -1)
170  edm::LogWarning("EvFDaqDirector") << "problem with creating filedesc for buwritelock -: "
171  << strerror(errno);
172  else
173  edm::LogInfo("EvFDaqDirector") << "creating filedesc for buwritelock -: "
174  << bu_writelock_fd_;
175  bu_w_lock_stream = fdopen(bu_writelock_fd_, "w");
176  if (bu_w_lock_stream == 0)
177  edm::LogWarning("EvFDaqDirector")<< "Error creating write lock stream -: " << strerror(errno);
178 
179  // BU INITIALIZES LOCK FILE
180  // FU LOCK FILE OPEN
181  openFULockfileStream(fulockfile, true);
183  fflush(fu_rw_lock_stream);
184  close(fu_readwritelock_fd_);
185 
186  if (hltSourceDirectory_.size())
187  {
188  struct stat buf;
189  if (stat(hltSourceDirectory_.c_str(),&buf)==0) {
190  std::string hltdir=bu_run_dir_+"/hlt";
191  std::string tmphltdir=bu_run_open_dir_+"/hlt";
192  retval = mkdir(tmphltdir.c_str(),S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
193 
194  boost::filesystem::copy_file(hltSourceDirectory_+"/HltConfig.py",tmphltdir+"/HltConfig.py");
195  try {
196  boost::filesystem::copy_file(hltSourceDirectory_+"/CMSSW_VERSION",tmphltdir+"/CMSSW_VERSION");
197  boost::filesystem::copy_file(hltSourceDirectory_+"/SCRAM_ARCH",tmphltdir+"/SCRAM_ARCH");
198  } catch (...) {}
199 
200  boost::filesystem::copy_file(hltSourceDirectory_+"/fffParameters.jsn",tmphltdir+"/fffParameters.jsn");
201 
202  boost::filesystem::rename(tmphltdir,hltdir);
203  }
204  else
205  throw cms::Exception("DaqDirector") << " Error looking for HLT configuration -: " << hltSourceDirectory_;
206  }
207  //else{}//no configuration specified
208  }
209  else
210  {
211  // for FU, check if bu base dir exists
212 
213  retval = mkdir(bu_base_dir_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
214  if (retval != 0 && errno != EEXIST) {
215  throw cms::Exception("DaqDirector") << " Error checking for bu base dir -: "
216  << bu_base_dir_ << " mkdir error:" << strerror(errno) << "\n";
217  }
218 
220  std::string fulockfile = bu_run_dir_ + "/fu.lock";
221  openFULockfileStream(fulockfile, false);
222  }
223 
224  pthread_mutex_init(&init_lock_,NULL);
225 
226  stopFilePath_ = run_dir_+"/CMSSW_STOP";
227  }
228 
230  {
231  if (fulocal_rwlock_fd_!=-1) {
232  unlockFULocal();
233  close(fulocal_rwlock_fd_);
234  }
235 
236  if (fulocal_rwlock_fd2_!=-1) {
237  unlockFULocal2();
238  close(fulocal_rwlock_fd2_);
239  }
240 
241  }
242 
243  void EvFDaqDirector::postEndRun(edm::GlobalContext const& globalContext) {
244  close(bu_readlock_fd_);
245  close(bu_writelock_fd_);
246  if (directorBu_) {
247  std::string filename = bu_run_dir_ + "/bu.lock";
248  removeFile(filename);
249  }
250  }
251 
253 
254  for (unsigned int i=0;i<bounds.maxNumberOfStreams();i++){
255  streamFileTracker_.push_back(-1);
256  }
257  nThreads_=bounds.maxNumberOfStreams();
258  nStreams_=bounds.maxNumberOfThreads();
259 
261  }
262 
263  void EvFDaqDirector::preBeginRun(edm::GlobalContext const& globalContext) {
264 
265  //assert(run_ == id.run());
266 
267  // check if the requested run is the latest one - issue a warning if it isn't
269  edm::LogWarning("EvFDaqDirector") << "WARNING - checking run dir -: "
270  << run_dir_ << ". This is not the highest run "
272  }
273  }
274 
276  {
277  //delete all files belonging to just closed lumi
278  unsigned int ls = globalContext.luminosityBlockID().luminosityBlock();
280  edm::LogWarning("EvFDaqDirector") << " Handles to check for files to delete were not set by the input source...";
281  return;
282  }
283 
284  std::unique_lock<std::mutex> lkw(*fileDeleteLockPtr_);
285  auto it = filesToDeletePtr_->begin();
286  while (it!=filesToDeletePtr_->end()) {
287  if (it->second->lumi_ == ls) {
288  const boost::filesystem::path filePath(it->second->fileName_);
289  LogDebug("EvFDaqDirector") << "Deleting input file -:" << it->second->fileName_;
290  try {
291  //rarely this fails but file gets deleted
292  boost::filesystem::remove(filePath);
293  }
294  catch (const boost::filesystem::filesystem_error& ex)
295  {
296  edm::LogError("EvFDaqDirector") << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() << ". Trying again.";
297  usleep(10000);
298  try {
299  boost::filesystem::remove(filePath);
300  }
301  catch (...) {/*file gets deleted first time but exception is still thrown*/}
302  }
303  catch (std::exception& ex)
304  {
305  edm::LogError("EvFDaqDirector") << " - deleteFile std::exception CAUGHT -: " << ex.what() << ". Trying again.";
306  usleep(10000);
307  try {
308  boost::filesystem::remove(filePath);
309  } catch (...) {/*file gets deleted first time but exception is still thrown*/}
310  }
311 
312  delete it->second;
313  it = filesToDeletePtr_->erase(it);
314  }
315  else it++;
316  }
317  }
318 
319  inline void EvFDaqDirector::preSourceEvent(edm::StreamID const& streamID) {
321  }
322 
323 
324  std::string EvFDaqDirector::getInputJsonFilePath(const unsigned int ls, const unsigned int index) const {
325  return bu_run_dir_ + "/" + fffnaming::inputJsonFileName(run_,ls,index);
326  }
327 
328 
329  std::string EvFDaqDirector::getRawFilePath(const unsigned int ls, const unsigned int index) const {
330  return bu_run_dir_ + "/" + fffnaming::inputRawFileName(run_,ls,index);
331  }
332 
333  std::string EvFDaqDirector::getOpenRawFilePath(const unsigned int ls, const unsigned int index) const {
334  return bu_run_dir_ + "/open/" + fffnaming::inputRawFileName(run_,ls,index);
335  }
336 
337  std::string EvFDaqDirector::getOpenInputJsonFilePath(const unsigned int ls, const unsigned int index) const {
338  return bu_run_dir_ + "/open/" + fffnaming::inputJsonFileName(run_,ls,index);
339  }
340 
342  return run_dir_ + "/open/" + fffnaming::streamerDataFileNameWithPid(run_,ls,stream);
343  }
344 
346  return run_dir_ + "/open/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
347  }
348 
350  return run_dir_ + "/" + fffnaming::streamerJsonFileNameWithPid(run_,ls,stream);
351  }
352 
355  }
356 
359  }
360 
362  return run_dir_ + "/open/" + fffnaming::initFileNameWithPid(run_,0,stream);
363  }
364 
366  return run_dir_ + "/" + fffnaming::initFileNameWithPid(run_,0,stream);
367  }
368 
370  return run_dir_ + "/open/" + fffnaming::protocolBufferHistogramFileNameWithPid(run_,ls,stream);
371  }
372 
375  }
376 
379  }
380 
382  return run_dir_ + "/open/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
383  }
384 
386  return run_dir_ + "/" + fffnaming::rootHistogramFileNameWithPid(run_,ls,stream);
387  }
388 
391  }
392 
394  return bu_run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
395  }
396 
398  return run_dir_ + "/" + fffnaming::eolsFileName(run_,ls);
399  }
400 
402  return run_dir_ + "/" + fffnaming::bolsFileName(run_,ls);
403  }
404 
406  return bu_run_dir_ + "/" + fffnaming::eorFileName(run_);
407  }
408 
409 
411  return run_dir_ + "/" + fffnaming::eorFileName(run_);
412  }
413 
415  int retval = remove(filename.c_str());
416  if (retval != 0)
417  edm::LogError("EvFDaqDirector") << "Could not remove used file -: " << filename << ". error = "
418  << strerror(errno);
419  }
420 
421  void EvFDaqDirector::removeFile(unsigned int ls, unsigned int index) {
422  removeFile(getRawFilePath(ls,index));
423  }
424 
425  EvFDaqDirector::FileStatus EvFDaqDirector::updateFuLock(unsigned int& ls, std::string& nextFile, uint32_t& fsize) {
426  EvFDaqDirector::FileStatus fileStatus = noFile;
427 
428  int retval = -1;
429  int lock_attempts = 0;
430 
431  struct stat buf;
432  int stopFileLS = -1;
433  if (stat(stopFilePath_.c_str(),&buf)==0) {
434  stopFileLS = readLastLSEntry(stopFilePath_);
435  edm::LogWarning("EvFDaqDirector") << "Detected stop request from hltd. Ending run for this process after LS -: " << stopFileLS;
436  //return runEnded;
437  }
438 
439  while (retval==-1) {
440  retval = fcntl(fu_readwritelock_fd_, F_SETLK, &fu_rw_flk);
441  if (retval==-1) usleep(50000);
442  else continue;
443 
444  lock_attempts++;
445  if (lock_attempts>100 || errno==116) {
446  if (errno==116)
447  edm::LogWarning("EvFDaqDirector") << "Stale lock file handle. Checking if run directory and fu.lock file are present" << std::endl;
448  else
449  edm::LogWarning("EvFDaqDirector") << "Unable to obtain a lock for 5 seconds. Checking if run directory and fu.lock file are present -: errno "
450  << errno <<":"<< strerror(errno) << std::endl;
451 
452  if (stat(bu_run_dir_.c_str(), &buf)!=0) return runEnded;
453  if (stat((bu_run_dir_+"/fu.lock").c_str(), &buf)!=0) return runEnded;
454  lock_attempts=0;
455  }
456  }
457  if(retval!=0) return fileStatus;
458 
459 #ifdef DEBUG
460  timeval ts_lockend;
461  gettimeofday(&ts_lockend,0);
462 #endif
463 
464  // if the stream is readable
465  if (fu_rw_lock_stream != 0) {
466  unsigned int readLs, readIndex, jumpLs, jumpIndex;
467  int check = 0;
468  // rewind the stream
469  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
470  // if rewinded ok
471  if (check == 0) {
472  // read its' values
474  fscanf(fu_rw_lock_stream, "%u %u %u %u", &readLs, &readIndex,
475  &jumpLs, &jumpIndex);
476  else {
477  fscanf(fu_rw_lock_stream, "%u %u", &readLs, &readIndex);
478  edm::LogInfo("EvFDaqDirector") << "Read fu.lock file file -: " << readLs << ":" << readIndex;
479  }
480 
481  // try to bump
482  bool bumpedOk = bumpFile(readLs, readIndex, nextFile, fsize, stopFileLS);
483  ls = readLs;
484  // there is a new file to grab or lumisection ended
485  if (bumpedOk) {
486  // rewind and clear
487  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
488  if (check == 0) {
489  ftruncate(fu_readwritelock_fd_, 0);
490  fflush(fu_rw_lock_stream); //this should not be needed ???
491  } else
492  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
493  << strerror(errno);
494  // write new data
495  check = fseek(fu_rw_lock_stream, 0, SEEK_SET);
496  if (check == 0) {
497  // write next index in the file, which is the file the next process should take
499  fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs,
500  readIndex + 1, readLs + 2, readIndex + 1);
501  jumpLS_ = readLs + 2;
503  } else {
504  fprintf(fu_rw_lock_stream, "%u %u", readLs,
505  readIndex + 1);
506  }
507  fflush(fu_rw_lock_stream);
508  fsync(fu_readwritelock_fd_);
509 
510  fileStatus = newFile;
511 
513  edm::LogInfo("EvFDaqDirector") << "Written to file -: " << readLs << ":"
514  << readIndex + 1 << " --> " << readLs + 2
515  << ":" << readIndex + 1;
516  else
517  LogDebug("EvFDaqDirector") << "Written to file -: " << readLs << ":"
518  << readIndex + 1;
519 
520  } else
521  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for updating failed with error "
522  << strerror(errno);
523  }
524  } else
525  edm::LogError("EvFDaqDirector") << "seek on fu read/write lock for reading failed with error "
526  << strerror(errno);
527  } else
528  edm::LogError("EvFDaqDirector") << "fu read/write lock stream is invalid " << strerror(errno);
529 
530 #ifdef DEBUG
531  timeval ts_preunlock;
532  gettimeofday(&ts_preunlock,0);
533  int locked_period_int = ts_preunlock.tv_sec - ts_lockend.tv_sec;
534  double locked_period=locked_period_int+double(ts_preunlock.tv_usec - ts_lockend.tv_usec)/1000000;
535 #endif
536 
537  //if new json is present, lock file which FedRawDataInputSource will later unlock
538  if (fileStatus==newFile) lockFULocal();
539 
540  //release lock at this point
541  int retvalu=-1;
542  retvalu=fcntl(fu_readwritelock_fd_, F_SETLKW, &fu_rw_fulk);
543  if (retvalu==-1) edm::LogError("EvFDaqDirector") << "Error unlocking the fu.lock " << strerror(errno);
544 
545 #ifdef DEBUG
546  edm::LogDebug("EvFDaqDirector") << "Waited during lock -: " << locked_period << " seconds";
547 #endif
548 
549  if ( fileStatus == noFile ) {
550  struct stat buf;
551  //edm::LogInfo("EvFDaqDirector") << " looking for EoR file: " << getEoRFilePath().c_str();
552  if ( stat(getEoRFilePath().c_str(), &buf) == 0 || stat(bu_run_dir_.c_str(), &buf)!=0)
553  fileStatus = runEnded;
554  if (stopFileLS>=0 && (int)ls > stopFileLS) {
555  edm::LogInfo("EvFDaqDirector") << "Reached maximum lumisection set by hltd";
556  fileStatus = runEnded;
557  }
558  }
559  return fileStatus;
560  }
561 
563 
564  boost::filesystem::ifstream ij(BUEoLSFile);
565  Json::Value deserializeRoot;
567 
568  if (!reader.parse(ij, deserializeRoot)) {
569  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << BUEoLSFile;
570  return -1;
571  }
572 
574  DataPoint dp;
575  dp.deserialize(deserializeRoot);
576 
577  //read definition
578  if (readEolsDefinition_) {
579  //std::string def = boost::algorithm::trim(dp.getDefinition());
581  if (!def.size()) readEolsDefinition_=false;
582  while (def.size()) {
583  std::string fullpath;
584  if (def.find('/')==0)
585  fullpath = def;
586  else
587  fullpath = bu_run_dir_+'/'+def;
588  struct stat buf;
589  if (stat(fullpath.c_str(), &buf) == 0) {
590  DataPointDefinition eolsDpd;
591  std::string defLabel = "legend";
592  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
593  if (eolsDpd.getNames().size()==0) {
594  //try with "data" label if "legend" format is not used
595  eolsDpd = DataPointDefinition();
596  defLabel="data";
597  DataPointDefinition::getDataPointDefinitionFor(fullpath, &eolsDpd,&defLabel);
598  }
599  for (unsigned int i=0;i<eolsDpd.getNames().size();i++)
600  if (eolsDpd.getNames().at(i)=="NFiles")
602  readEolsDefinition_=false;
603  break;
604  }
605  //check if we can still find definition
606  if (def.size()<=1 || def.find('/')==std::string::npos) {
607  readEolsDefinition_=false;
608  break;
609  }
610  def = def.substr(def.find('/')+1);
611  }
612  }
613 
614  if (dp.getData().size()>eolsNFilesIndex_)
615  data = dp.getData()[eolsNFilesIndex_];
616  else {
617  edm::LogError("EvFDaqDirector") << " error reading number of files from BU JSON -: " << BUEoLSFile;
618  return -1;
619  }
620  return boost::lexical_cast<int>(data);
621  }
622 
623  bool EvFDaqDirector::bumpFile(unsigned int& ls, unsigned int& index, std::string& nextFile, uint32_t& fsize, int maxLS) {
624 
625  if (previousFileSize_ != 0) {
626  if (!fms_) {
627  try {
629  } catch (...) {
630  edm::LogError("EvFDaqDirector") <<" FastMonitoringService not found";
631  }
632  }
634  previousFileSize_ = 0;
635  }
636 
637  struct stat buf;
638  std::stringstream ss;
639  unsigned int nextIndex = index;
640  nextIndex++;
641 
642  // 1. Check suggested file
643  nextFile = getInputJsonFilePath(ls,index);
644  if (stat(nextFile.c_str(), &buf) == 0) {
645 
646  previousFileSize_ = buf.st_size;
647  fsize = buf.st_size;
648  return true;
649  }
650  // 2. No file -> lumi ended? (and how many?)
651  else {
652  std::string BUEoLSFile = getEoLSFilePathOnBU(ls);
653  bool eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
654  unsigned int startingLumi = ls;
655  while (eolFound) {
656 
657  // recheck that no raw file appeared in the meantime
658  if (stat(nextFile.c_str(), &buf) == 0) {
659  previousFileSize_ = buf.st_size;
660  fsize = buf.st_size;
661  return true;
662  }
663 
664  int indexFilesInLS = getNFilesFromEoLS(BUEoLSFile);
665  if (indexFilesInLS < 0)
666  //parsing failed
667  return false;
668  else {
669  //check index
670  if ((int)index<indexFilesInLS) {
671  //we have 2 files, and check for 1 failed... retry (2 will never be here)
672  edm::LogError("EvFDaqDirector") << "Potential miss of index file in LS -: " << ls << ". Missing "
673  << nextFile << " because " << indexFilesInLS-1 << " is the highest index expected. Will not update fu.lock file";
674  return false;
675  }
676  }
677  // this lumi ended, check for files
678  ++ls;
679  index = 0;
680 
681  //reached limit
682  if (maxLS>=0 && ls > (unsigned int)maxLS) return false;
683 
684  nextFile = getInputJsonFilePath(ls,0);
685  if (stat(nextFile.c_str(), &buf) == 0) {
686  // a new file was found at new lumisection, index 0
687  previousFileSize_ = buf.st_size;
688  fsize = buf.st_size;
689 
691  // rename ended lumi to + 2
692  std::string sourceEol = getEoLSFilePathOnBU(startingLumi);
693 
694  std::string destEol = getEoLSFilePathOnBU(startingLumi+2);
695 
696  std::string cpCmd = "cp " + sourceEol + " " + destEol;
697  edm::LogInfo("EvFDaqDirector") << " testmode: Running copy cmd -: " << cpCmd;
698  int rc = system(cpCmd.c_str());
699  if (rc != 0) {
700  edm::LogError("EvFDaqDirector") << " testmode: COPY EOL FAILED!!!!! -: " << cpCmd;
701  }
702  }
703 
704  return true;
705  }
706  BUEoLSFile = getEoLSFilePathOnBU(ls);
707  eolFound = (stat(BUEoLSFile.c_str(), &buf) == 0);
708  }
709  }
710  // no new file found
711  return false;
712  }
713 
715  if (fu_rw_lock_stream == 0)
716  edm::LogError("EvFDaqDirector") << "Error creating fu read/write lock stream "
717  << strerror(errno);
718  else {
719  edm::LogInfo("EvFDaqDirector") << "Initializing FU LOCK FILE";
720  unsigned int readLs = 1, readIndex = 0, jumpLs = 3, jumpIndex = 0;
722  fprintf(fu_rw_lock_stream, "%u %u %u %u", readLs, readIndex,
723  jumpLs, jumpIndex);
724  else
725  fprintf(fu_rw_lock_stream, "%u %u", readLs, readIndex);
726  }
727  }
728 
730  if (create) {
731  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR | O_CREAT,
732  S_IRWXU | S_IWGRP | S_IRGRP | S_IWOTH | S_IROTH);
733  chmod(fulockfile.c_str(),0766);
734  } else {
735  fu_readwritelock_fd_ = open(fulockfile.c_str(), O_RDWR, S_IRWXU);
736  }
737  if (fu_readwritelock_fd_ == -1)
738  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for fuwritelock -: " << fulockfile.c_str()
739  << " create:" << create << " error:" << strerror(errno);
740  else
741  LogDebug("EvFDaqDirector") << "creating filedesc for fureadwritelock -: "
743 
744  fu_rw_lock_stream = fdopen(fu_readwritelock_fd_, "r+");
745  }
746 
747  //create if does not exist then lock the merge destination file
749  data_rw_stream = fopen(getMergedDatFilePath(ls,stream).c_str(), "a"); //open stream for appending
751  if (data_readwrite_fd_ == -1)
752  edm::LogError("EvFDaqDirector") << "problem with creating filedesc for datamerge "
753  << strerror(errno);
754  else
755  LogDebug("EvFDaqDirector") << "creating filedesc for datamerge -: "
757  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_flk);
758 
759  return data_rw_stream;
760  }
761 
763  fflush(data_rw_stream);
764  fcntl(data_readwrite_fd_, F_SETLKW, &data_rw_fulk);
765  fclose(data_rw_stream);
766  }
767 
769  pthread_mutex_lock(&init_lock_);
770  }
771 
773  pthread_mutex_unlock(&init_lock_);
774  }
775 
777  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_flk);
778  flock(fulocal_rwlock_fd_,LOCK_EX);
779  }
780 
782  //fcntl(fulocal_rwlock_fd_, F_SETLKW, &fulocal_rw_fulk);
783  flock(fulocal_rwlock_fd_,LOCK_UN);
784  }
785 
786 
788  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_flk2);
789  flock(fulocal_rwlock_fd2_,LOCK_EX);
790  }
791 
793  //fcntl(fulocal_rwlock_fd2_, F_SETLKW, &fulocal_rw_fulk2);
794  flock(fulocal_rwlock_fd2_,LOCK_UN);
795  }
796 
797 
799  // create open dir if not already there
800 
802  if (!boost::filesystem::is_directory(openPath)) {
803  LogDebug("EvFDaqDirector") << "<open> FU dir not found. Creating... -:" << openPath.string();
804  boost::filesystem::create_directories(openPath);
805  }
806  }
807 
808 
810 
811  boost::filesystem::ifstream ij(file);
812  Json::Value deserializeRoot;
814 
815  if (!reader.parse(ij, deserializeRoot)) {
816  edm::LogError("EvFDaqDirector") << "Cannot deserialize input JSON file -:" << file;
817  return -1;
818  }
819 
820  int ret = deserializeRoot.get("lastLS","").asInt();
821  return ret;
822 
823  }
824 
825  //if transferSystem PSet is present in the menu, we require it to be complete and consistent for all specified streams
827  {
828  transferSystemJson_.reset(new Json::Value);
829  if (edm::getProcessParameterSet().existsAs<edm::ParameterSet>("transferSystem",true))
830  {
831  const edm::ParameterSet& tsPset(edm::getProcessParameterSet().getParameterSet("transferSystem"));
832 
833  Json::Value destinationsVal(Json::arrayValue);
834  std::vector<std::string> destinations = tsPset.getParameter<std::vector<std::string>>("destinations");
835  for (auto & dest: destinations) destinationsVal.append(dest);
836  (*transferSystemJson_)["destinations"]=destinationsVal;
837 
838  Json::Value modesVal(Json::arrayValue);
839  std::vector<std::string> modes = tsPset.getParameter< std::vector<std::string> >("transferModes");
840  for (auto & mode: modes) modesVal.append(mode);
841  (*transferSystemJson_)["transferModes"]=modesVal;
842 
843  for (auto psKeyItr =tsPset.psetTable().begin();psKeyItr!=tsPset.psetTable().end(); ++ psKeyItr) {
844  if (psKeyItr->first!="destinations" && psKeyItr->first!="transferModes") {
845  const edm::ParameterSet & streamDef = tsPset.getParameterSet(psKeyItr->first);
846  Json::Value streamVal;
847  for (auto & mode : modes) {
848  //validation
849  if (!streamDef.existsAs<std::vector<std::string>>(mode,true))
850  throw cms::Exception("EvFDaqDirector") << " Missing transfer system specification for -:" << psKeyItr->first << " (transferMode " << mode << ")";
851  std::vector<std::string> streamDestinations = streamDef.getParameter<std::vector<std::string>>(mode);
852 
853  Json::Value sDestsValue(Json::arrayValue);
854 
855  if (!streamDestinations.size())
856  throw cms::Exception("EvFDaqDirector") << " Missing transter system destination(s) for -: "<< psKeyItr->first << ", mode:" << mode;
857 
858  for (auto & sdest:streamDestinations) {
859  bool sDestValid=false;
860  sDestsValue.append(sdest);
861  for (auto & dest: destinations) {
862  if (dest==sdest) sDestValid=true;
863  }
864  if (!sDestValid)
865  throw cms::Exception("EvFDaqDirector") << " Invalid transter system destination specified for -: "<< psKeyItr->first << ", mode:" << mode << ", dest:"<<sdest;
866  }
867  streamVal[mode]=sDestsValue;
868  }
869  (*transferSystemJson_)[psKeyItr->first] = streamVal;
870  }
871  }
872  }
873  else {
874  if (requireTSPSet_)
875  throw cms::Exception("EvFDaqDirector") << "transferSystem PSet not found";
876  }
877  }
878 
880  {
881  std::string streamRequestName;
882  if (transferSystemJson_->isMember(stream.c_str()))
883  streamRequestName = stream;
884  else {
885  std::stringstream msg;
886  msg << "Transfer system mode definitions missing for -: " << stream;
887  if (requireTSPSet_)
888  throw cms::Exception("EvFDaqDirector") << msg.str();
889  else {
890  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
891  return std::string();
892  }
893  }
894  //return empty if strict check parameter is not on
896  edm::LogWarning("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter."
897  << "Switch on requireTSPSet parameter to enforce this requirement. Setting mode to empty string.";
898  return std::string();
899  }
901  throw cms::Exception("EvFDaqDirector") << "Selected mode string is not provided as DaqDirector parameter.";
902  }
903  //check if stream has properly listed transfer stream
904  if (!transferSystemJson_->get(streamRequestName, "").isMember(selectedTransferMode_.c_str()))
905  {
906  std::stringstream msg;
907  msg << "Selected transfer mode " << selectedTransferMode_ << " is not specified for stream " << streamRequestName;
908  if (requireTSPSet_)
909  throw cms::Exception("EvFDaqDirector") << msg.str();
910  else
911  edm::LogWarning("EvFDaqDirector") << msg.str() << " (permissive mode)";
912  return std::string();
913  }
914  Json::Value destsVec = transferSystemJson_->get(streamRequestName, "").get(selectedTransferMode_,"");
915 
916  //flatten string json::Array into CSV std::string
918  for (Json::Value::iterator it = destsVec.begin(); it!=destsVec.end(); it++)
919  {
920  if (ret!="") ret +=",";
921  ret+=(*it).asString();
922  }
923  return ret;
924  }
925 
926 }
#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:185
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_
FileStatus updateFuLock(unsigned int &ls, std::string &nextFile, uint32_t &fsize)
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 &)
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:256
Int asInt() const
void preSourceEvent(edm::StreamID const &streamID)
std::string initFileNameWithPid(const unsigned int run, const unsigned int ls, std::string const &stream)
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
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: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)
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