CMS 3D CMS Logo

All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Modules Pages
FedRawDataInputSource.cc
Go to the documentation of this file.
1 #include <fcntl.h>
2 #include <iomanip>
3 #include <iostream>
4 #include <memory>
5 #include <sstream>
6 #include <sys/types.h>
7 #include <sys/file.h>
8 #include <sys/time.h>
9 #include <unistd.h>
10 #include <vector>
11 #include <fstream>
12 #include <zlib.h>
13 #include <cstdio>
14 #include <chrono>
15 
16 #include <boost/algorithm/string.hpp>
17 
21 
23 
30 
32 
34 
38 
40 
45 
46 //JSON file reader
48 
49 #include <boost/lexical_cast.hpp>
50 
51 using namespace evf::FastMonState;
52 
54  : edm::RawInputSource(pset, desc),
55  defPath_(pset.getUntrackedParameter<std::string>("buDefPath", "")),
56  eventChunkSize_(pset.getUntrackedParameter<unsigned int>("eventChunkSize", 32) * 1048576),
57  eventChunkBlock_(pset.getUntrackedParameter<unsigned int>("eventChunkBlock", 32) * 1048576),
58  numBuffers_(pset.getUntrackedParameter<unsigned int>("numBuffers", 2)),
59  maxBufferedFiles_(pset.getUntrackedParameter<unsigned int>("maxBufferedFiles", 2)),
60  getLSFromFilename_(pset.getUntrackedParameter<bool>("getLSFromFilename", true)),
61  alwaysStartFromFirstLS_(pset.getUntrackedParameter<bool>("alwaysStartFromFirstLS", false)),
62  verifyChecksum_(pset.getUntrackedParameter<bool>("verifyChecksum", true)),
63  useL1EventID_(pset.getUntrackedParameter<bool>("useL1EventID", false)),
64  testTCDSFEDRange_(
65  pset.getUntrackedParameter<std::vector<unsigned int>>("testTCDSFEDRange", std::vector<unsigned int>())),
66  fileNames_(pset.getUntrackedParameter<std::vector<std::string>>("fileNames", std::vector<std::string>())),
67  fileListMode_(pset.getUntrackedParameter<bool>("fileListMode", false)),
68  fileListLoopMode_(pset.getUntrackedParameter<bool>("fileListLoopMode", false)),
69  runNumber_(edm::Service<evf::EvFDaqDirector>()->getRunNumber()),
70  daqProvenanceHelper_(edm::TypeID(typeid(FEDRawDataCollection))),
71  eventID_(),
72  processHistoryID_(),
73  currentLumiSection_(0),
74  tcds_pointer_(nullptr),
75  eventsThisLumi_(0) {
76  char thishost[256];
77  gethostname(thishost, 255);
78  edm::LogInfo("FedRawDataInputSource") << "Construction. read-ahead chunk size -: " << std::endl
79  << (eventChunkSize_ / 1048576) << " MB on host " << thishost;
80 
81  if (!testTCDSFEDRange_.empty()) {
82  if (testTCDSFEDRange_.size() != 2) {
83  throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection")
84  << "Invalid TCDS Test FED range parameter";
85  }
88  }
89 
90  long autoRunNumber = -1;
91  if (fileListMode_) {
92  autoRunNumber = initFileList();
93  if (!fileListLoopMode_) {
94  if (autoRunNumber < 0)
95  throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource") << "Run number not found from filename";
96  //override run number
97  runNumber_ = (edm::RunNumber_t)autoRunNumber;
98  edm::Service<evf::EvFDaqDirector>()->overrideRunNumber((unsigned int)autoRunNumber);
99  }
100  }
101 
103  setNewRun();
104  //todo:autodetect from file name (assert if names differ)
106 
107  //make sure that chunk size is N * block size
112 
113  if (!numBuffers_)
114  throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource")
115  << "no reading enabled with numBuffers parameter 0";
116 
119  readingFilesCount_ = 0;
120 
121  if (!crc32c_hw_test())
122  edm::LogError("FedRawDataInputSource::FedRawDataInputSource") << "Intel crc32c checksum computation unavailable";
123 
124  //get handles to DaqDirector and FastMonitoringService because getting them isn't possible in readSupervisor thread
125  if (fileListMode_) {
126  try {
127  fms_ = static_cast<evf::FastMonitoringService*>(edm::Service<evf::MicroStateService>().operator->());
128  } catch (cms::Exception const&) {
129  edm::LogInfo("FedRawDataInputSource") << "No FastMonitoringService found in the configuration";
130  }
131  } else {
132  fms_ = static_cast<evf::FastMonitoringService*>(edm::Service<evf::MicroStateService>().operator->());
133  if (!fms_) {
134  throw cms::Exception("FedRawDataInputSource") << "FastMonitoringService not found";
135  }
136  }
137 
139  if (!daqDirector_)
140  cms::Exception("FedRawDataInputSource") << "EvFDaqDirector not found";
141 
143  if (useFileBroker_)
144  edm::LogInfo("FedRawDataInputSource") << "EvFDaqDirector/Source configured to use file service";
145  //set DaqDirector to delete files in preGlobalEndLumi callback
147  if (fms_) {
149  fms_->setInputSource(this);
152  }
153  //should delete chunks when run stops
154  for (unsigned int i = 0; i < numBuffers_; i++) {
156  }
157 
158  quit_threads_ = false;
159 
160  for (unsigned int i = 0; i < numConcurrentReads_; i++) {
161  std::unique_lock<std::mutex> lk(startupLock_);
162  //issue a memory fence here and in threads (constructor was segfaulting without this)
163  thread_quit_signal.push_back(false);
164  workerJob_.push_back(ReaderInfo(nullptr, nullptr));
165  cvReader_.push_back(new std::condition_variable);
166  tid_active_.push_back(0);
167  threadInit_.store(false, std::memory_order_release);
168  workerThreads_.push_back(new std::thread(&FedRawDataInputSource::readWorker, this, i));
169  startupCv_.wait(lk);
170  }
171 
172  runAuxiliary()->setProcessHistoryID(processHistoryID_);
173 }
174 
176  quit_threads_ = true;
177 
178  //delete any remaining open files
179  for (auto it = filesToDelete_.begin(); it != filesToDelete_.end(); it++)
180  it->second.reset();
181 
183  readSupervisorThread_->join();
184  } else {
185  //join aux threads in case the supervisor thread was not started
186  for (unsigned int i = 0; i < workerThreads_.size(); i++) {
187  std::unique_lock<std::mutex> lk(mReader_);
188  thread_quit_signal[i] = true;
189  cvReader_[i]->notify_one();
190  lk.unlock();
191  workerThreads_[i]->join();
192  delete workerThreads_[i];
193  }
194  }
195  for (unsigned int i = 0; i < numConcurrentReads_; i++)
196  delete cvReader_[i];
197  /*
198  for (unsigned int i=0;i<numConcurrentReads_+1;i++) {
199  InputChunk *ch;
200  while (!freeChunks_.try_pop(ch)) {}
201  delete ch;
202  }
203  */
204 }
205 
208  desc.setComment("File-based Filter Farm input source for reading raw data from BU ramdisk");
209  desc.addUntracked<unsigned int>("eventChunkSize", 32)->setComment("Input buffer (chunk) size");
210  desc.addUntracked<unsigned int>("eventChunkBlock", 32)
211  ->setComment("Block size used in a single file read call (must be smaller or equal to buffer size)");
212  desc.addUntracked<unsigned int>("numBuffers", 2)->setComment("Number of buffers used for reading input");
213  desc.addUntracked<unsigned int>("maxBufferedFiles", 2)
214  ->setComment("Maximum number of simultaneously buffered raw files");
215  desc.addUntracked<unsigned int>("alwaysStartFromfirstLS", false)
216  ->setComment("Force source to start from LS 1 if server provides higher lumisection number");
217  desc.addUntracked<bool>("verifyChecksum", true)
218  ->setComment("Verify event CRC-32C checksum of FRDv5 and higher or Adler32 with v3 and v4");
219  desc.addUntracked<bool>("useL1EventID", false)
220  ->setComment("Use L1 event ID from FED header if true or from TCDS FED if false");
221  desc.addUntracked<std::vector<unsigned int>>("testTCDSFEDRange", std::vector<unsigned int>())
222  ->setComment("[min, max] range to search for TCDS FED ID in test setup");
223  desc.addUntracked<bool>("fileListMode", false)
224  ->setComment("Use fileNames parameter to directly specify raw files to open");
225  desc.addUntracked<std::vector<std::string>>("fileNames", std::vector<std::string>())
226  ->setComment("file list used when fileListMode is enabled");
227  desc.setAllowAnything();
228  descriptions.add("source", desc);
229 }
230 
233  //this thread opens new files and dispatches reading to worker readers
234  //threadInit_.store(false,std::memory_order_release);
235  std::unique_lock<std::mutex> lk(startupLock_);
236  readSupervisorThread_ = std::make_unique<std::thread>(&FedRawDataInputSource::readSupervisor, this);
238  startupCv_.wait(lk);
239  }
240  //signal hltd to start event accounting
241  if (!currentLumiSection_)
244  switch (nextEvent()) {
246  //maybe create EoL file in working directory before ending run
247  struct stat buf;
248  if (!useFileBroker_ && currentLumiSection_ > 0) {
249  bool eolFound = (stat(daqDirector_->getEoLSFilePathOnBU(currentLumiSection_).c_str(), &buf) == 0);
250  if (eolFound) {
252  bool found = (stat(fuEoLS.c_str(), &buf) == 0);
253  if (!found) {
255  int eol_fd =
256  open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
257  close(eol_fd);
259  }
260  }
261  }
262  //also create EoR file in FU data directory
263  bool eorFound = (stat(daqDirector_->getEoRFilePathOnFU().c_str(), &buf) == 0);
264  if (!eorFound) {
265  int eor_fd = open(daqDirector_->getEoRFilePathOnFU().c_str(),
266  O_RDWR | O_CREAT,
267  S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
268  close(eor_fd);
269  }
271  eventsThisLumi_ = 0;
273  edm::LogInfo("FedRawDataInputSource") << "----------------RUN ENDED----------------";
274  return Next::kStop;
275  }
277  //this is not reachable
278  return Next::kEvent;
279  }
281  //std::cout << "--------------NEW LUMI---------------" << std::endl;
282  return Next::kEvent;
283  }
284  default: {
285  if (!getLSFromFilename_) {
286  //get new lumi from file header
287  if (event_->lumi() > currentLumiSection_) {
289  eventsThisLumi_ = 0;
291  }
292  }
295  else
296  eventRunNumber_ = event_->run();
297  L1EventID_ = event_->event();
298 
299  setEventCached();
300 
301  return Next::kEvent;
302  }
303  }
304 }
305 
306 void FedRawDataInputSource::maybeOpenNewLumiSection(const uint32_t lumiSection) {
307  if (!luminosityBlockAuxiliary() || luminosityBlockAuxiliary()->luminosityBlock() != lumiSection) {
308  if (!useFileBroker_) {
309  if (currentLumiSection_ > 0) {
311  struct stat buf;
312  bool found = (stat(fuEoLS.c_str(), &buf) == 0);
313  if (!found) {
315  int eol_fd =
316  open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
317  close(eol_fd);
318  daqDirector_->createBoLSFile(lumiSection, false);
320  }
321  } else
322  daqDirector_->createBoLSFile(lumiSection, true); //needed for initial lumisection
323  }
324 
325  currentLumiSection_ = lumiSection;
326 
328 
329  timeval tv;
330  gettimeofday(&tv, nullptr);
331  const edm::Timestamp lsopentime((unsigned long long)tv.tv_sec * 1000000 + (unsigned long long)tv.tv_usec);
332 
334  runAuxiliary()->run(), lumiSection, lsopentime, edm::Timestamp::invalidTimestamp());
335 
336  setLuminosityBlockAuxiliary(lumiBlockAuxiliary);
337  luminosityBlockAuxiliary()->setProcessHistoryID(processHistoryID_);
338 
339  edm::LogInfo("FedRawDataInputSource") << "New lumi section was opened. LUMI -: " << lumiSection;
340  }
341 }
342 
346  if (edm::shutdown_flag.load(std::memory_order_relaxed))
347  break;
348  }
349  return status;
350 }
351 
353  if (setExceptionState_)
354  threadError();
355  if (!currentFile_.get()) {
358  if (!fileQueue_.try_pop(currentFile_)) {
359  //sleep until wakeup (only in single-buffer mode) or timeout
360  std::unique_lock<std::mutex> lkw(mWakeup_);
361  if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout || !currentFile_.get())
363  }
364  status = currentFile_->status_;
367  currentFile_.reset();
368  return status;
369  } else if (status == evf::EvFDaqDirector::runAbort) {
370  throw cms::Exception("FedRawDataInputSource::getNextEvent")
371  << "Run has been aborted by the input source reader thread";
372  } else if (status == evf::EvFDaqDirector::newLumi) {
374  if (getLSFromFilename_) {
375  if (currentFile_->lumi_ > currentLumiSection_) {
377  eventsThisLumi_ = 0;
379  }
380  } else { //let this be picked up from next event
382  }
383  currentFile_.reset();
384  return status;
385  } else if (status == evf::EvFDaqDirector::newFile) {
387  } else
388  assert(false);
389  }
391 
392  //file is empty
393  if (!currentFile_->fileSize_) {
395  //try to open new lumi
396  assert(currentFile_->nChunks_ == 0);
397  if (getLSFromFilename_)
398  if (currentFile_->lumi_ > currentLumiSection_) {
400  eventsThisLumi_ = 0;
402  }
403  //immediately delete empty file
404  currentFile_.reset();
406  }
407 
408  //file is finished
409  if (currentFile_->bufferPosition_ == currentFile_->fileSize_) {
411  //release last chunk (it is never released elsewhere)
412  freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_]);
413  if (currentFile_->nEvents_ >= 0 && currentFile_->nEvents_ != int(currentFile_->nProcessed_)) {
414  throw cms::Exception("FedRawDataInputSource::getNextEvent")
415  << "Fully processed " << currentFile_->nProcessed_ << " from the file " << currentFile_->fileName_
416  << " but according to BU JSON there should be " << currentFile_->nEvents_ << " events";
417  }
418  //try to wake up supervisor thread which might be sleeping waiting for the free chunk
419  if (singleBufferMode_) {
420  std::unique_lock<std::mutex> lkw(mWakeup_);
421  cvWakeup_.notify_one();
422  }
423  bufferInputRead_ = 0;
425  //put the file in pending delete list;
426  std::unique_lock<std::mutex> lkw(fileDeleteLock_);
427  filesToDelete_.push_back(std::pair<int, std::unique_ptr<InputFile>>(currentFileIndex_, std::move(currentFile_)));
428  } else {
429  //in single-thread and stream jobs, events are already processed
430  currentFile_.reset();
431  }
433  }
434 
435  //handle RAW file header
436  if (currentFile_->bufferPosition_ == 0 && currentFile_->rawHeaderSize_ > 0) {
437  if (currentFile_->fileSize_ <= currentFile_->rawHeaderSize_) {
438  if (currentFile_->fileSize_ < currentFile_->rawHeaderSize_)
439  throw cms::Exception("FedRawDataInputSource::getNextEvent")
440  << "Premature end of input file while reading file header";
441 
442  edm::LogWarning("FedRawDataInputSource")
443  << "File with only raw header and no events received in LS " << currentFile_->lumi_;
444  if (getLSFromFilename_)
445  if (currentFile_->lumi_ > currentLumiSection_) {
447  eventsThisLumi_ = 0;
449  }
450  }
451 
452  //advance buffer position to skip file header (chunk will be acquired later)
453  currentFile_->chunkPosition_ += currentFile_->rawHeaderSize_;
454  currentFile_->bufferPosition_ += currentFile_->rawHeaderSize_;
455  }
456 
457  //file is too short
458  if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < FRDHeaderVersionSize[detectedFRDversion_]) {
459  throw cms::Exception("FedRawDataInputSource::getNextEvent")
460  << "Premature end of input file while reading event header";
461  }
462  if (singleBufferMode_) {
463  //should already be there
465  while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
466  usleep(10000);
467  if (currentFile_->parent_->exceptionState() || setExceptionState_)
468  currentFile_->parent_->threadError();
469  }
471 
472  unsigned char* dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
473 
474  //conditions when read amount is not sufficient for the header to fit
478 
479  if (detectedFRDversion_ == 0) {
480  detectedFRDversion_ = *((uint16_t*)dataPosition);
482  throw cms::Exception("FedRawDataInputSource::getNextEvent")
483  << "Unknown FRD version -: " << detectedFRDversion_;
485  }
486 
487  //recalculate chunk position
488  dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
490  throw cms::Exception("FedRawDataInputSource::getNextEvent")
491  << "Premature end of input file while reading event header";
492  }
493  }
494 
495  event_ = std::make_unique<FRDEventMsgView>(dataPosition);
496  if (event_->size() > eventChunkSize_) {
497  throw cms::Exception("FedRawDataInputSource::getNextEvent")
498  << " event id:" << event_->event() << " lumi:" << event_->lumi() << " run:" << event_->run()
499  << " of size:" << event_->size() << " bytes does not fit into a chunk of size:" << eventChunkSize_
500  << " bytes";
501  }
502 
503  const uint32_t msgSize = event_->size() - FRDHeaderVersionSize[detectedFRDversion_];
504 
505  if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize) {
506  throw cms::Exception("FedRawDataInputSource::getNextEvent")
507  << "Premature end of input file while reading event data";
508  }
509  if (eventChunkSize_ - currentFile_->chunkPosition_ < msgSize) {
511  //recalculate chunk position
512  dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
513  event_ = std::make_unique<FRDEventMsgView>(dataPosition);
514  }
515  currentFile_->bufferPosition_ += event_->size();
516  currentFile_->chunkPosition_ += event_->size();
517  //last chunk is released when this function is invoked next time
518 
519  }
520  //multibuffer mode:
521  else {
522  //wait for the current chunk to become added to the vector
524  while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
525  usleep(10000);
526  if (setExceptionState_)
527  threadError();
528  }
530 
531  //check if header is at the boundary of two chunks
532  chunkIsFree_ = false;
533  unsigned char* dataPosition;
534 
535  //read header, copy it to a single chunk if necessary
536  bool chunkEnd = currentFile_->advance(dataPosition, FRDHeaderVersionSize[detectedFRDversion_]);
537 
538  event_ = std::make_unique<FRDEventMsgView>(dataPosition);
539  if (event_->size() > eventChunkSize_) {
540  throw cms::Exception("FedRawDataInputSource::getNextEvent")
541  << " event id:" << event_->event() << " lumi:" << event_->lumi() << " run:" << event_->run()
542  << " of size:" << event_->size() << " bytes does not fit into a chunk of size:" << eventChunkSize_
543  << " bytes";
544  }
545 
546  const uint32_t msgSize = event_->size() - FRDHeaderVersionSize[detectedFRDversion_];
547 
548  if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize) {
549  throw cms::Exception("FedRawDataInputSource::getNextEvent")
550  << "Premature end of input file while reading event data";
551  }
552 
553  if (chunkEnd) {
554  //header was at the chunk boundary, we will have to move payload as well
555  currentFile_->moveToPreviousChunk(msgSize, FRDHeaderVersionSize[detectedFRDversion_]);
556  chunkIsFree_ = true;
557  } else {
558  //header was contiguous, but check if payload fits the chunk
559  if (eventChunkSize_ - currentFile_->chunkPosition_ < msgSize) {
560  //rewind to header start position
562  //copy event to a chunk start and move pointers
563 
565 
566  chunkEnd = currentFile_->advance(dataPosition, FRDHeaderVersionSize[detectedFRDversion_] + msgSize);
567 
569 
570  assert(chunkEnd);
571  chunkIsFree_ = true;
572  //header is moved
573  event_ = std::make_unique<FRDEventMsgView>(dataPosition);
574  } else {
575  //everything is in a single chunk, only move pointers forward
576  chunkEnd = currentFile_->advance(dataPosition, msgSize);
577  assert(!chunkEnd);
578  chunkIsFree_ = false;
579  }
580  }
581  } //end multibuffer mode
583 
584  if (verifyChecksum_ && event_->version() >= 5) {
585  uint32_t crc = 0;
586  crc = crc32c(crc, (const unsigned char*)event_->payload(), event_->eventSize());
587  if (crc != event_->crc32c()) {
588  if (fms_)
590  throw cms::Exception("FedRawDataInputSource::getNextEvent")
591  << "Found a wrong crc32c checksum: expected 0x" << std::hex << event_->crc32c() << " but calculated 0x"
592  << crc;
593  }
594  } else if (verifyChecksum_ && event_->version() >= 3) {
595  uint32_t adler = adler32(0L, Z_NULL, 0);
596  adler = adler32(adler, (Bytef*)event_->payload(), event_->eventSize());
597 
598  if (adler != event_->adler32()) {
599  if (fms_)
601  throw cms::Exception("FedRawDataInputSource::getNextEvent")
602  << "Found a wrong Adler32 checksum: expected 0x" << std::hex << event_->adler32() << " but calculated 0x"
603  << adler;
604  }
605  }
607 
608  currentFile_->nProcessed_++;
609 
611 }
612 
615  std::unique_ptr<FEDRawDataCollection> rawData(new FEDRawDataCollection);
616  bool tcdsInRange;
617  edm::Timestamp tstamp = fillFEDRawDataCollection(*rawData, tcdsInRange);
618 
619  if (useL1EventID_) {
622  aux.setProcessHistoryID(processHistoryID_);
623  makeEvent(eventPrincipal, aux);
624  } else if (tcds_pointer_ == nullptr) {
625  if (!GTPEventID_) {
626  throw cms::Exception("FedRawDataInputSource::read")
627  << "No TCDS or GTP FED in event with FEDHeader EID -: " << L1EventID_;
628  }
631  aux.setProcessHistoryID(processHistoryID_);
632  makeEvent(eventPrincipal, aux);
633  } else {
634  const FEDHeader fedHeader(tcds_pointer_);
635  tcds::Raw_v1 const* tcds = reinterpret_cast<tcds::Raw_v1 const*>(tcds_pointer_ + FEDHeader::length);
640  event_->isRealData(),
641  static_cast<edm::EventAuxiliary::ExperimentType>(fedHeader.triggerType()),
642  processGUID(),
644  !tcdsInRange);
645  aux.setProcessHistoryID(processHistoryID_);
646  makeEvent(eventPrincipal, aux);
647  }
648 
649  std::unique_ptr<edm::WrapperBase> edp(new edm::Wrapper<FEDRawDataCollection>(std::move(rawData)));
650 
652 
653  eventsThisLumi_++;
655 
656  //resize vector if needed
657  while (streamFileTracker_.size() <= eventPrincipal.streamID())
658  streamFileTracker_.push_back(-1);
659 
660  streamFileTracker_[eventPrincipal.streamID()] = currentFileIndex_;
661 
662  //this old file check runs no more often than every 10 events
663  if (!((currentFile_->nProcessed_ - 1) % (checkEvery_))) {
664  //delete files that are not in processing
665  std::unique_lock<std::mutex> lkw(fileDeleteLock_);
666  auto it = filesToDelete_.begin();
667  while (it != filesToDelete_.end()) {
668  bool fileIsBeingProcessed = false;
669  for (unsigned int i = 0; i < nStreams_; i++) {
670  if (it->first == streamFileTracker_.at(i)) {
671  fileIsBeingProcessed = true;
672  break;
673  }
674  }
675  if (!fileIsBeingProcessed) {
676  std::string fileToDelete = it->second->fileName_;
677  it = filesToDelete_.erase(it);
678  } else
679  it++;
680  }
681  }
682  if (chunkIsFree_)
683  freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_ - 1]);
684  chunkIsFree_ = false;
686  return;
687 }
688 
691  timeval stv;
692  gettimeofday(&stv, nullptr);
693  time = stv.tv_sec;
694  time = (time << 32) + stv.tv_usec;
695  edm::Timestamp tstamp(time);
696 
697  uint32_t eventSize = event_->eventSize();
698  unsigned char* event = (unsigned char*)event_->payload();
699  GTPEventID_ = 0;
700  tcds_pointer_ = nullptr;
701  tcdsInRange = false;
702  uint16_t selectedTCDSFed = 0;
703  while (eventSize > 0) {
704  assert(eventSize >= FEDTrailer::length);
705  eventSize -= FEDTrailer::length;
706  const FEDTrailer fedTrailer(event + eventSize);
707  const uint32_t fedSize = fedTrailer.fragmentLength() << 3; //trailer length counts in 8 bytes
708  assert(eventSize >= fedSize - FEDHeader::length);
709  eventSize -= (fedSize - FEDHeader::length);
710  const FEDHeader fedHeader(event + eventSize);
711  const uint16_t fedId = fedHeader.sourceID();
713  throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection") << "Out of range FED ID : " << fedId;
714  } else if (fedId >= MINTCDSuTCAFEDID_ && fedId <= MAXTCDSuTCAFEDID_) {
715  if (!selectedTCDSFed) {
716  selectedTCDSFed = fedId;
717  tcds_pointer_ = event + eventSize;
719  tcdsInRange = true;
720  }
721  } else
722  throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection")
723  << "Second TCDS FED ID " << fedId << " found. First ID: " << selectedTCDSFed;
724  }
726  if (evf::evtn::evm_board_sense(event + eventSize, fedSize))
727  GTPEventID_ = evf::evtn::get(event + eventSize, true);
728  else
729  GTPEventID_ = evf::evtn::get(event + eventSize, false);
730  //evf::evtn::evm_board_setformat(fedSize);
731  const uint64_t gpsl = evf::evtn::getgpslow(event + eventSize);
732  const uint64_t gpsh = evf::evtn::getgpshigh(event + eventSize);
733  tstamp = edm::Timestamp(static_cast<edm::TimeValue_t>((gpsh << 32) + gpsl));
734  }
735  //take event ID from GTPE FED
737  if (evf::evtn::gtpe_board_sense(event + eventSize)) {
738  GTPEventID_ = evf::evtn::gtpe_get(event + eventSize);
739  }
740  }
741  FEDRawData& fedData = rawData.FEDData(fedId);
742  fedData.resize(fedSize);
743  memcpy(fedData.data(), event + eventSize, fedSize);
744  }
745  assert(eventSize == 0);
746 
747  return tstamp;
748 }
749 
751 
753  bool stop = false;
754  unsigned int currentLumiSection = 0;
755  //threadInit_.exchange(true,std::memory_order_acquire);
756 
757  {
758  std::unique_lock<std::mutex> lk(startupLock_);
759  startupCv_.notify_one();
760  }
761 
762  uint32_t ls = 0;
763  uint32_t monLS = 1;
764  uint32_t lockCount = 0;
765  uint64_t sumLockWaitTimeUs = 0.;
766 
767  while (!stop) {
768  //wait for at least one free thread and chunk
769  int counter = 0;
770  while ((workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty() ||
772  //report state to monitoring
773  if (fms_) {
774  bool copy_active = false;
775  for (auto j : tid_active_)
776  if (j)
777  copy_active = true;
780  else if (freeChunks_.empty()) {
781  if (copy_active)
783  else
785  } else {
786  if (copy_active)
788  else
790  }
791  }
792  std::unique_lock<std::mutex> lkw(mWakeup_);
793  //sleep until woken up by condition or a timeout
794  if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout) {
795  counter++;
796  //if (!(counter%50)) edm::LogInfo("FedRawDataInputSource") << "No free chunks or threads...";
797  LogDebug("FedRawDataInputSource") << "No free chunks or threads...";
798  } else {
799  assert(!(workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty());
800  }
801  if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
802  stop = true;
803  break;
804  }
805  }
806  //if this is reached, there are enough buffers and threads to proceed or processing is instructed to stop
807 
808  if (stop)
809  break;
810 
811  //look for a new file
812  std::string nextFile;
813  uint32_t fileSizeIndex;
814  int64_t fileSizeFromMetadata;
815 
816  if (fms_) {
819  }
820 
822  uint16_t rawHeaderSize = 0;
823  uint32_t lsFromRaw = 0;
824  int32_t serverEventsInNewFile = -1;
825  int rawFd = -1;
826 
827  int backoff_exp = 0;
828 
829  //entering loop which tries to grab new file from ramdisk
831  if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
832  stop = true;
833  break;
834  }
835 
836  assert(rawFd == -1);
837  uint64_t thisLockWaitTimeUs = 0.;
839  if (fileListMode_) {
840  //return LS if LS not set, otherwise return file
841  status = getFile(ls, nextFile, fileSizeIndex, thisLockWaitTimeUs);
844  rawFd,
845  rawHeaderSize,
846  lsFromRaw,
847  serverEventsInNewFile,
848  fileSizeFromMetadata,
849  false,
850  false,
851  false) != 0) {
852  //error
853  setExceptionState_ = true;
854  stop = true;
855  break;
856  }
857  if (!getLSFromFilename_)
858  ls = lsFromRaw;
859  }
860  } else if (!useFileBroker_)
862  ls, nextFile, fileSizeIndex, rawHeaderSize, thisLockWaitTimeUs, setExceptionState_);
863  else {
864  status = daqDirector_->getNextFromFileBroker(currentLumiSection,
865  ls,
866  nextFile,
867  rawFd,
868  rawHeaderSize,
869  serverEventsInNewFile,
870  fileSizeFromMetadata,
871  thisLockWaitTimeUs);
872  }
873 
875 
876  //cycle through all remaining LS even if no files get assigned
877  if (currentLumiSection != ls && status == evf::EvFDaqDirector::runEnded)
879 
880  //monitoring of lock wait time
881  if (thisLockWaitTimeUs > 0.)
882  sumLockWaitTimeUs += thisLockWaitTimeUs;
883  lockCount++;
884  if (ls > monLS) {
885  monLS = ls;
886  if (lockCount)
887  if (fms_)
888  fms_->reportLockWait(monLS, sumLockWaitTimeUs, lockCount);
889  lockCount = 0;
890  sumLockWaitTimeUs = 0;
891  }
892 
893  //check again for any remaining index/EoLS files after EoR file is seen
896  usleep(100000);
897  //now all files should have appeared in ramdisk, check again if any raw files were left behind
899  ls, nextFile, fileSizeIndex, rawHeaderSize, thisLockWaitTimeUs, setExceptionState_);
900  if (currentLumiSection != ls && status == evf::EvFDaqDirector::runEnded)
902  }
903 
905  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runEnded));
906  fileQueue_.push(std::move(inf));
907  stop = true;
908  break;
909  }
910 
911  //error from filelocking function
913  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runAbort, 0));
914  fileQueue_.push(std::move(inf));
915  stop = true;
916  break;
917  }
918  //queue new lumisection
919  if (getLSFromFilename_) {
920  if (ls > currentLumiSection) {
921  if (!useFileBroker_) {
922  //file locking
923  //setMonStateSup(inSupNewLumi);
924  currentLumiSection = ls;
925  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, currentLumiSection));
926  fileQueue_.push(std::move(inf));
927  } else {
928  //new file service
929  if (currentLumiSection == 0 && !alwaysStartFromFirstLS_) {
930  if (ls < 100) {
931  //look at last LS file on disk to start from that lumisection (only within first 100 LS)
932  unsigned int lsToStart = daqDirector_->getLumisectionToStart();
933 
934  for (unsigned int nextLS = lsToStart; nextLS <= ls; nextLS++) {
935  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, nextLS));
936  fileQueue_.push(std::move(inf));
937  }
938  } else {
939  //start from current LS
940  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, ls));
941  fileQueue_.push(std::move(inf));
942  }
943  } else {
944  //queue all lumisections after last one seen to avoid gaps
945  for (unsigned int nextLS = currentLumiSection + 1; nextLS <= ls; nextLS++) {
946  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, nextLS));
947  fileQueue_.push(std::move(inf));
948  }
949  }
950  currentLumiSection = ls;
951  }
952  }
953  //else
954  if (currentLumiSection > 0 && ls < currentLumiSection) {
955  edm::LogError("FedRawDataInputSource")
956  << "Got old LS (" << ls << ") file from EvFDAQDirector! Expected LS:" << currentLumiSection
957  << ". Aborting execution." << std::endl;
958  if (rawFd != -1)
959  close(rawFd);
960  rawFd = -1;
961  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runAbort, 0));
962  fileQueue_.push(std::move(inf));
963  stop = true;
964  break;
965  }
966  }
967 
968  int dbgcount = 0;
971  dbgcount++;
972  if (!(dbgcount % 20))
973  LogDebug("FedRawDataInputSource") << "No file for me... sleep and try again...";
974  if (!useFileBroker_)
975  usleep(100000);
976  else {
977  backoff_exp = std::min(4, backoff_exp); // max 1.6 seconds
978  //backoff_exp=0; // disabled!
979  int sleeptime = (int)(100000. * pow(2, backoff_exp));
980  usleep(sleeptime);
981  backoff_exp++;
982  }
983  } else
984  backoff_exp = 0;
985  }
986  //end of file grab loop, parse result
989  LogDebug("FedRawDataInputSource") << "The director says to grab -: " << nextFile;
990 
991  std::string rawFile;
992  //file service will report raw extension
993  if (useFileBroker_ || rawHeaderSize)
994  rawFile = nextFile;
995  else {
996  std::filesystem::path rawFilePath(nextFile);
997  rawFile = rawFilePath.replace_extension(".raw").string();
998  }
999 
1000  struct stat st;
1001  int stat_res = stat(rawFile.c_str(), &st);
1002  if (stat_res == -1) {
1003  edm::LogError("FedRawDataInputSource") << "Can not stat file (" << errno << "):-" << rawFile << std::endl;
1004  setExceptionState_ = true;
1005  break;
1006  }
1007  uint64_t fileSize = st.st_size;
1008 
1009  if (fms_) {
1013  }
1014  int eventsInNewFile;
1015  if (fileListMode_) {
1016  if (fileSize == 0)
1017  eventsInNewFile = 0;
1018  else
1019  eventsInNewFile = -1;
1020  } else {
1022  if (!useFileBroker_) {
1023  if (rawHeaderSize) {
1024  int rawFdEmpty = -1;
1025  uint16_t rawHeaderCheck;
1026  bool fileFound;
1027  eventsInNewFile = daqDirector_->grabNextJsonFromRaw(
1028  nextFile, rawFdEmpty, rawHeaderCheck, fileSizeFromMetadata, fileFound, 0, true);
1029  assert(fileFound && rawHeaderCheck == rawHeaderSize);
1031  } else
1032  eventsInNewFile = daqDirector_->grabNextJsonFileAndUnlock(nextFile);
1033  } else
1034  eventsInNewFile = serverEventsInNewFile;
1035  assert(eventsInNewFile >= 0);
1036  assert((eventsInNewFile > 0) ==
1037  (fileSize > rawHeaderSize)); //file without events must be empty or contain only header
1038  }
1039 
1040  if (!singleBufferMode_) {
1041  //calculate number of needed chunks
1042  unsigned int neededChunks = fileSize / eventChunkSize_;
1043  if (fileSize % eventChunkSize_)
1044  neededChunks++;
1045 
1046  std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1047  ls,
1048  rawFile,
1049  !fileListMode_,
1050  rawFd,
1051  fileSize,
1052  rawHeaderSize,
1053  neededChunks,
1054  eventsInNewFile,
1055  this));
1057  auto newInputFilePtr = newInputFile.get();
1058  fileQueue_.push(std::move(newInputFile));
1059 
1060  for (unsigned int i = 0; i < neededChunks; i++) {
1061  if (fms_) {
1062  bool copy_active = false;
1063  for (auto j : tid_active_)
1064  if (j)
1065  copy_active = true;
1066  if (copy_active)
1068  else
1070  }
1071  //get thread
1072  unsigned int newTid = 0xffffffff;
1073  while (!workerPool_.try_pop(newTid)) {
1074  usleep(100000);
1075  if (quit_threads_.load(std::memory_order_relaxed)) {
1076  stop = true;
1077  break;
1078  }
1079  }
1080 
1081  if (fms_) {
1082  bool copy_active = false;
1083  for (auto j : tid_active_)
1084  if (j)
1085  copy_active = true;
1086  if (copy_active)
1088  else
1090  }
1091  InputChunk* newChunk = nullptr;
1092  while (!freeChunks_.try_pop(newChunk)) {
1093  usleep(100000);
1094  if (quit_threads_.load(std::memory_order_relaxed)) {
1095  stop = true;
1096  break;
1097  }
1098  }
1099 
1100  if (newChunk == nullptr) {
1101  //return unused tid if we received shutdown (nullptr chunk)
1102  if (newTid != 0xffffffff)
1103  workerPool_.push(newTid);
1104  stop = true;
1105  break;
1106  }
1107  if (stop)
1108  break;
1110 
1111  std::unique_lock<std::mutex> lk(mReader_);
1112 
1113  unsigned int toRead = eventChunkSize_;
1114  if (i == neededChunks - 1 && fileSize % eventChunkSize_)
1115  toRead = fileSize % eventChunkSize_;
1116  newChunk->reset(i * eventChunkSize_, toRead, i);
1117 
1118  workerJob_[newTid].first = newInputFilePtr;
1119  workerJob_[newTid].second = newChunk;
1120 
1121  //wake up the worker thread
1122  cvReader_[newTid]->notify_one();
1123  }
1124  } else {
1125  if (!eventsInNewFile) {
1126  if (rawFd) {
1127  close(rawFd);
1128  rawFd = -1;
1129  }
1130  //still queue file for lumi update
1131  std::unique_lock<std::mutex> lkw(mWakeup_);
1132  //TODO: also file with only file header fits in this edge case. Check if read correctly in single buffer mode
1133  std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1134  ls,
1135  rawFile,
1136  !fileListMode_,
1137  rawFd,
1138  fileSize,
1139  rawHeaderSize,
1140  (rawHeaderSize > 0),
1141  0,
1142  this));
1144  fileQueue_.push(std::move(newInputFile));
1145  cvWakeup_.notify_one();
1146  break;
1147  }
1148  //in single-buffer mode put single chunk in the file and let the main thread read the file
1149  InputChunk* newChunk;
1150  //should be available immediately
1151  while (!freeChunks_.try_pop(newChunk)) {
1152  usleep(100000);
1153  if (quit_threads_.load(std::memory_order_relaxed))
1154  break;
1155  }
1156 
1157  std::unique_lock<std::mutex> lkw(mWakeup_);
1158 
1159  unsigned int toRead = eventChunkSize_;
1160  if (fileSize % eventChunkSize_)
1161  toRead = fileSize % eventChunkSize_;
1162  newChunk->reset(0, toRead, 0);
1163  newChunk->readComplete_ = true;
1164 
1165  //push file and wakeup main thread
1166  std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1167  ls,
1168  rawFile,
1169  !fileListMode_,
1170  rawFd,
1171  fileSize,
1172  rawHeaderSize,
1173  1,
1174  eventsInNewFile,
1175  this));
1176  newInputFile->chunks_[0] = newChunk;
1178  fileQueue_.push(std::move(newInputFile));
1179  cvWakeup_.notify_one();
1180  }
1181  }
1182  }
1184  //make sure threads finish reading
1185  unsigned numFinishedThreads = 0;
1186  while (numFinishedThreads < workerThreads_.size()) {
1187  unsigned tid = 0;
1188  while (!workerPool_.try_pop(tid)) {
1189  usleep(10000);
1190  }
1191  std::unique_lock<std::mutex> lk(mReader_);
1192  thread_quit_signal[tid] = true;
1193  cvReader_[tid]->notify_one();
1194  numFinishedThreads++;
1195  }
1196  for (unsigned int i = 0; i < workerThreads_.size(); i++) {
1197  workerThreads_[i]->join();
1198  delete workerThreads_[i];
1199  }
1200 }
1201 
1202 void FedRawDataInputSource::readWorker(unsigned int tid) {
1203  bool init = true;
1204  threadInit_.exchange(true, std::memory_order_acquire);
1205 
1206  while (true) {
1207  tid_active_[tid] = false;
1208  std::unique_lock<std::mutex> lk(mReader_);
1209  workerJob_[tid].first = nullptr;
1210  workerJob_[tid].first = nullptr;
1211 
1212  assert(!thread_quit_signal[tid]); //should never get it here
1213  workerPool_.push(tid);
1214 
1215  if (init) {
1216  std::unique_lock<std::mutex> lk(startupLock_);
1217  init = false;
1218  startupCv_.notify_one();
1219  }
1220  cvReader_[tid]->wait(lk);
1221 
1222  if (thread_quit_signal[tid])
1223  return;
1224  tid_active_[tid] = true;
1225 
1226  InputFile* file;
1227  InputChunk* chunk;
1228 
1229  assert(workerJob_[tid].first != nullptr && workerJob_[tid].second != nullptr);
1230 
1231  file = workerJob_[tid].first;
1232  chunk = workerJob_[tid].second;
1233 
1234  //skip reading initial header size in first chunk if inheriting file descriptor (already set at appropriate position)
1235  unsigned int bufferLeft = (chunk->offset_ == 0 && file->rawFd_ != -1) ? file->rawHeaderSize_ : 0;
1236 
1237  //if only one worker thread exists, use single fd for all operations
1238  //if more worker threads exist, use rawFd_ for only the first read operation and then close file
1239  int fileDescriptor;
1240  bool fileOpenedHere = false;
1241 
1242  if (numConcurrentReads_ == 1) {
1243  fileDescriptor = file->rawFd_;
1244  if (fileDescriptor == -1) {
1245  fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
1246  fileOpenedHere = true;
1247  file->rawFd_ = fileDescriptor;
1248  }
1249  } else {
1250  if (chunk->offset_ == 0) {
1251  fileDescriptor = file->rawFd_;
1252  file->rawFd_ = -1;
1253  if (fileDescriptor == -1) {
1254  fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
1255  fileOpenedHere = true;
1256  }
1257  } else {
1258  fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
1259  fileOpenedHere = true;
1260  }
1261  }
1262 
1263  if (fileDescriptor < 0) {
1264  edm::LogError("FedRawDataInputSource") << "readWorker failed to open file -: " << file->fileName_
1265  << " fd:" << fileDescriptor << " error: " << strerror(errno);
1266  setExceptionState_ = true;
1267  continue;
1268  }
1269  if (fileOpenedHere) { //fast forward to this chunk position
1270  off_t pos = 0;
1271  pos = lseek(fileDescriptor, chunk->offset_, SEEK_SET);
1272  if (pos == -1) {
1273  edm::LogError("FedRawDataInputSource")
1274  << "readWorker failed to seek file -: " << file->fileName_ << " fd:" << fileDescriptor << " to offset "
1275  << chunk->offset_ << " error: " << strerror(errno);
1276  setExceptionState_ = true;
1277  continue;
1278  }
1279  }
1280 
1281  LogDebug("FedRawDataInputSource") << "Reader thread opened file -: TID: " << tid << " file: " << file->fileName_
1282  << " at offset " << lseek(fileDescriptor, 0, SEEK_CUR);
1283 
1284  unsigned int skipped = bufferLeft;
1286  for (unsigned int i = 0; i < readBlocks_; i++) {
1287  ssize_t last;
1288 
1289  //protect against reading into next block
1290  last = ::read(
1291  fileDescriptor, (void*)(chunk->buf_ + bufferLeft), std::min(chunk->usedSize_ - bufferLeft, eventChunkBlock_));
1292 
1293  if (last < 0) {
1294  edm::LogError("FedRawDataInputSource") << "readWorker failed to read file -: " << file->fileName_
1295  << " fd:" << fileDescriptor << " error: " << strerror(errno);
1296  setExceptionState_ = true;
1297  break;
1298  }
1299  if (last > 0)
1300  bufferLeft += last;
1301  if (last < eventChunkBlock_) { //last read
1302  //check if this is last block, then total read size must match file size
1303  if (!(chunk->usedSize_ - skipped == i * eventChunkBlock_ + last)) {
1304  edm::LogError("FedRawDataInputSource")
1305  << "readWorker failed to read file -: " << file->fileName_ << " fd:" << fileDescriptor << " last:" << last
1306  << " expectedChunkSize:" << chunk->usedSize_
1307  << " readChunkSize:" << (skipped + i * eventChunkBlock_ + last) << " skipped:" << skipped
1308  << " block:" << (i + 1) << "/" << readBlocks_ << " error: " << strerror(errno);
1309  setExceptionState_ = true;
1310  }
1311  break;
1312  }
1313  }
1314  if (setExceptionState_)
1315  continue;
1316 
1318  auto diff = end - start;
1319  std::chrono::milliseconds msec = std::chrono::duration_cast<std::chrono::milliseconds>(diff);
1320  LogDebug("FedRawDataInputSource") << " finished reading block -: " << (bufferLeft >> 20) << " MB"
1321  << " in " << msec.count() << " ms (" << (bufferLeft >> 20) / double(msec.count())
1322  << " GB/s)";
1323 
1324  if (chunk->offset_ + bufferLeft == file->fileSize_) { //file reading finished using same fd
1325  close(fileDescriptor);
1326  fileDescriptor = -1;
1327  if (numConcurrentReads_ == 1)
1328  file->rawFd_ = -1;
1329  }
1330  if (numConcurrentReads_ > 1 && fileDescriptor != -1)
1331  close(fileDescriptor);
1332 
1333  //detect FRD event version. Skip file Header if it exists
1334  if (detectedFRDversion_ == 0 && chunk->offset_ == 0) {
1335  detectedFRDversion_ = *((uint16_t*)(chunk->buf_ + file->rawHeaderSize_));
1336  }
1338  chunk->readComplete_ =
1339  true; //this is atomic to secure the sequential buffer fill before becoming available for processing)
1340  file->chunks_[chunk->fileIndex_] = chunk; //put the completed chunk in the file chunk vector at predetermined index
1341  }
1342 }
1343 
1345  quit_threads_ = true;
1346  throw cms::Exception("FedRawDataInputSource:threadError") << " file reader thread error ";
1347 }
1348 
1350  if (fms_)
1351  fms_->setInState(state);
1352 }
1353 
1355  if (fms_)
1357 }
1358 
1359 inline bool InputFile::advance(unsigned char*& dataPosition, const size_t size) {
1360  //wait for chunk
1361 
1362  while (!waitForChunk(currentChunk_)) {
1364  usleep(100000);
1366  if (parent_->exceptionState())
1367  parent_->threadError();
1368  }
1369 
1370  dataPosition = chunks_[currentChunk_]->buf_ + chunkPosition_;
1371  size_t currentLeft = chunks_[currentChunk_]->size_ - chunkPosition_;
1372 
1373  if (currentLeft < size) {
1374  //we need next chunk
1375  while (!waitForChunk(currentChunk_ + 1)) {
1377  usleep(100000);
1379  if (parent_->exceptionState())
1380  parent_->threadError();
1381  }
1382  //copy everything to beginning of the first chunk
1383  dataPosition -= chunkPosition_;
1384  assert(dataPosition == chunks_[currentChunk_]->buf_);
1385  memmove(chunks_[currentChunk_]->buf_, chunks_[currentChunk_]->buf_ + chunkPosition_, currentLeft);
1386  memcpy(chunks_[currentChunk_]->buf_ + currentLeft, chunks_[currentChunk_ + 1]->buf_, size - currentLeft);
1387  //set pointers at the end of the old data position
1388  bufferPosition_ += size;
1389  chunkPosition_ = size - currentLeft;
1390  currentChunk_++;
1391  return true;
1392  } else {
1393  chunkPosition_ += size;
1394  bufferPosition_ += size;
1395  return false;
1396  }
1397 }
1398 
1399 inline void InputFile::moveToPreviousChunk(const size_t size, const size_t offset) {
1400  //this will fail in case of events that are too large
1402  assert(size - offset < chunks_[currentChunk_]->size_);
1403  memcpy(chunks_[currentChunk_ - 1]->buf_ + offset, chunks_[currentChunk_]->buf_ + chunkPosition_, size);
1404  chunkPosition_ += size;
1405  bufferPosition_ += size;
1406 }
1407 
1408 inline void InputFile::rewindChunk(const size_t size) {
1409  chunkPosition_ -= size;
1410  bufferPosition_ -= size;
1411 }
1412 
1414  if (rawFd_ != -1)
1415  close(rawFd_);
1416 
1417  if (deleteFile_ && !fileName_.empty()) {
1419  try {
1420  //sometimes this fails but file gets deleted
1421  LogDebug("FedRawDataInputSource:InputFile") << "Deleting input file -:" << fileName_;
1423  return;
1424  } catch (const std::filesystem::filesystem_error& ex) {
1425  edm::LogError("FedRawDataInputSource:InputFile")
1426  << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() << ". Trying again.";
1427  } catch (std::exception& ex) {
1428  edm::LogError("FedRawDataInputSource:InputFile")
1429  << " - deleteFile std::exception CAUGHT -: " << ex.what() << ". Trying again.";
1430  }
1432  }
1433 }
1434 
1435 //single-buffer mode file reading
1437  uint32_t existingSize = 0;
1438 
1439  if (fileDescriptor_ < 0) {
1440  bufferInputRead_ = 0;
1441  if (file->rawFd_ == -1) {
1442  fileDescriptor_ = open(file->fileName_.c_str(), O_RDONLY);
1443  if (file->rawHeaderSize_)
1444  lseek(fileDescriptor_, file->rawHeaderSize_, SEEK_SET);
1445  } else
1446  fileDescriptor_ = file->rawFd_;
1447 
1448  //skip header size in destination buffer (chunk position was already adjusted)
1449  bufferInputRead_ += file->rawHeaderSize_;
1450  existingSize += file->rawHeaderSize_;
1451 
1452  if (fileDescriptor_ >= 0)
1453  LogDebug("FedRawDataInputSource") << "opened file -: " << std::endl << file->fileName_;
1454  else {
1455  throw cms::Exception("FedRawDataInputSource:readNextChunkIntoBuffer")
1456  << "failed to open file " << std::endl
1457  << file->fileName_ << " fd:" << fileDescriptor_;
1458  }
1459  //fill chunk (skipping file header if present)
1460  for (unsigned int i = 0; i < readBlocks_; i++) {
1461  const ssize_t last = ::read(fileDescriptor_,
1462  (void*)(file->chunks_[0]->buf_ + existingSize),
1463  eventChunkBlock_ - (i == readBlocks_ - 1 ? existingSize : 0));
1465  existingSize += last;
1466  }
1467 
1468  } else {
1469  //continue reading
1470  if (file->chunkPosition_ == 0) { //in the rare case the last byte barely fit
1471  for (unsigned int i = 0; i < readBlocks_; i++) {
1472  const ssize_t last = ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSize), eventChunkBlock_);
1474  existingSize += last;
1475  }
1476  } else {
1477  //event didn't fit in last chunk, so leftover must be moved to the beginning and completed
1478  uint32_t existingSizeLeft = eventChunkSize_ - file->chunkPosition_;
1479  memmove((void*)file->chunks_[0]->buf_, file->chunks_[0]->buf_ + file->chunkPosition_, existingSizeLeft);
1480 
1481  //calculate amount of data that can be added
1482  const uint32_t blockcount = file->chunkPosition_ / eventChunkBlock_;
1483  const uint32_t leftsize = file->chunkPosition_ % eventChunkBlock_;
1484 
1485  for (uint32_t i = 0; i < blockcount; i++) {
1486  const ssize_t last =
1487  ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSizeLeft), eventChunkBlock_);
1489  existingSizeLeft += last;
1490  }
1491  if (leftsize) {
1492  const ssize_t last = ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSizeLeft), leftsize);
1494  }
1495  file->chunkPosition_ = 0; //data was moved to beginning of the chunk
1496  }
1497  }
1498  if (bufferInputRead_ == file->fileSize_) { // no more data in this file
1499  if (fileDescriptor_ != -1) {
1500  LogDebug("FedRawDataInputSource") << "Closing input file -: " << std::endl << file->fileName_;
1501  close(fileDescriptor_);
1502  file->rawFd_ = fileDescriptor_ = -1;
1503  }
1504  }
1505 }
1506 
1508  std::lock_guard<std::mutex> lock(monlock_);
1509  auto itr = sourceEventsReport_.find(lumi);
1510  if (itr != sourceEventsReport_.end())
1511  itr->second += events;
1512  else
1514 }
1515 
1516 std::pair<bool, unsigned int> FedRawDataInputSource::getEventReport(unsigned int lumi, bool erase) {
1517  std::lock_guard<std::mutex> lock(monlock_);
1518  auto itr = sourceEventsReport_.find(lumi);
1519  if (itr != sourceEventsReport_.end()) {
1520  std::pair<bool, unsigned int> ret(true, itr->second);
1521  if (erase)
1522  sourceEventsReport_.erase(itr);
1523  return ret;
1524  } else
1525  return std::pair<bool, unsigned int>(false, 0);
1526 }
1527 
1529  std::sort(fileNames_.begin(), fileNames_.end(), [](std::string a, std::string b) {
1530  if (a.rfind('/') != std::string::npos)
1531  a = a.substr(a.rfind('/'));
1532  if (b.rfind('/') != std::string::npos)
1533  b = b.substr(b.rfind('/'));
1534  return b > a;
1535  });
1536 
1537  if (!fileNames_.empty()) {
1538  //get run number from first file in the vector
1540  std::string fileStem = fileName.stem().string();
1541  if (fileStem.find("file://") == 0)
1542  fileStem = fileStem.substr(7);
1543  else if (fileStem.find("file:") == 0)
1544  fileStem = fileStem.substr(5);
1545  auto end = fileStem.find('_');
1546 
1547  if (fileStem.find("run") == 0) {
1548  std::string runStr = fileStem.substr(3, end - 3);
1549  try {
1550  //get long to support test run numbers < 2^32
1551  long rval = boost::lexical_cast<long>(runStr);
1552  edm::LogInfo("FedRawDataInputSource") << "Autodetected run number in fileListMode -: " << rval;
1553  return rval;
1554  } catch (boost::bad_lexical_cast const&) {
1555  edm::LogWarning("FedRawDataInputSource")
1556  << "Unable to autodetect run number in fileListMode from file -: " << fileName;
1557  }
1558  }
1559  }
1560  return -1;
1561 }
1562 
1564  std::string& nextFile,
1565  uint32_t& fsize,
1566  uint64_t& lockWaitTime) {
1567  if (fileListIndex_ < fileNames_.size()) {
1568  nextFile = fileNames_[fileListIndex_];
1569  if (nextFile.find("file://") == 0)
1570  nextFile = nextFile.substr(7);
1571  else if (nextFile.find("file:") == 0)
1572  nextFile = nextFile.substr(5);
1573  std::filesystem::path fileName = nextFile;
1574  std::string fileStem = fileName.stem().string();
1575  if (fileStem.find("ls"))
1576  fileStem = fileStem.substr(fileStem.find("ls") + 2);
1577  if (fileStem.find('_'))
1578  fileStem = fileStem.substr(0, fileStem.find('_'));
1579 
1580  if (!fileListLoopMode_)
1581  ls = boost::lexical_cast<unsigned int>(fileStem);
1582  else //always starting from LS 1 in loop mode
1583  ls = 1 + loopModeIterationInc_;
1584 
1585  //fsize = 0;
1586  //lockWaitTime = 0;
1587  fileListIndex_++;
1589  } else {
1590  if (!fileListLoopMode_)
1592  else {
1593  //loop through files until interrupted
1595  fileListIndex_ = 0;
1596  return getFile(ls, nextFile, fsize, lockWaitTime);
1597  }
1598  }
1599 }
runTheMatrix.ret
ret
prodAgent to be discontinued
Definition: runTheMatrix.py:543
FedRawDataInputSource::fileDeleteLock_
std::mutex fileDeleteLock_
Definition: FedRawDataInputSource.h:167
change_name.diff
diff
Definition: change_name.py:13
evf::EvFDaqDirector::unlockFULocal2
void unlockFULocal2()
Definition: EvFDaqDirector.cc:928
evf::FastMonState::inSupWaitFreeThreadCopying
Definition: FastMonitoringService.h:107
eostools.ls
def ls(path, rec=False)
Definition: eostools.py:349
edm::RunNumber_t
unsigned int RunNumber_t
Definition: RunLumiEventNumber.h:14
counter
Definition: counter.py:1
FedRawDataInputSource::chunkIsFree_
bool chunkIsFree_
Definition: FedRawDataInputSource.h:142
dttmaxenums::L
Definition: DTTMax.h:29
evf::EvFDaqDirector::createBoLSFile
void createBoLSFile(const uint32_t lumiSection, bool checkIfExists) const
Definition: EvFDaqDirector.cc:933
FedRawDataInputSource::threadInit_
std::atomic< bool > threadInit_
Definition: FedRawDataInputSource.h:181
electrons_cff.bool
bool
Definition: electrons_cff.py:366
mps_fire.i
i
Definition: mps_fire.py:428
FedRawDataInputSource::rewind_
void rewind_() override
Definition: FedRawDataInputSource.cc:750
start
Definition: start.py:1
evf::evtn::makeEventAuxiliary
edm::EventAuxiliary makeEventAuxiliary(const tcds::Raw_v1 *, unsigned int runNumber, unsigned int lumiSection, bool isRealData, const edm::EventAuxiliary::ExperimentType &, const std::string &processGUID, bool verifyLumiSection, bool suppressWarning)
Definition: AuxiliaryMakers.cc:9
evf::EvFDaqDirector::runEnded
Definition: EvFDaqDirector.h:64
MessageLogger.h
evf::FastMonState::inSupNewFileWaitChunkCopying
Definition: FastMonitoringService.h:115
InputFile::parent_
FedRawDataInputSource * parent_
Definition: FedRawDataInputSource.h:212
evf::EvFDaqDirector::getEoLSFilePathOnBU
std::string getEoLSFilePathOnBU(const unsigned int ls) const
Definition: EvFDaqDirector.cc:476
funct::false
false
Definition: Factorize.h:29
FEDNumbering::MINTriggerEGTPFEDID
Definition: FEDNumbering.h:63
InputFile::fileName_
std::string fileName_
Definition: FedRawDataInputSource.h:215
FedRawDataInputSource::maybeOpenNewLumiSection
void maybeOpenNewLumiSection(const uint32_t lumiSection)
Definition: FedRawDataInputSource.cc:306
FedRawDataInputSource::quit_threads_
std::atomic< bool > quit_threads_
Definition: FedRawDataInputSource.h:158
FedRawDataInputSource::mWakeup_
std::mutex mWakeup_
Definition: FedRawDataInputSource.h:173
FedRawDataInputSource::fileNames_
std::vector< std::string > fileNames_
Definition: FedRawDataInputSource.h:102
evf::FastMonState::inWaitChunk
Definition: FastMonitoringService.h:92
FedRawDataInputSource::workerThreads_
std::vector< std::thread * > workerThreads_
Definition: FedRawDataInputSource.h:146
FedRawDataInputSource::setMonState
void setMonState(evf::FastMonState::InputState state)
Definition: FedRawDataInputSource.cc:1349
FedRawDataInputSource::fileListIndex_
unsigned int fileListIndex_
Definition: FedRawDataInputSource.h:107
FedRawDataInputSource::MINTCDSuTCAFEDID_
uint16_t MINTCDSuTCAFEDID_
Definition: FedRawDataInputSource.h:129
evf::FastMonState::inChunkReceived
Definition: FastMonitoringService.h:93
FedRawDataInputSource::threadError
void threadError()
Definition: FedRawDataInputSource.cc:1344
FedRawDataInputSource::readingFilesCount_
std::atomic< unsigned int > readingFilesCount_
Definition: FedRawDataInputSource.h:94
submitPVValidationJobs.now
now
Definition: submitPVValidationJobs.py:639
mps_update.status
status
Definition: mps_update.py:68
evf::FastMonState::inSupNewFileWaitThread
Definition: FastMonitoringService.h:114
FedRawDataInputSource::useL1EventID_
const bool useL1EventID_
Definition: FedRawDataInputSource.h:100
min
T min(T a, T b)
Definition: MathUtil.h:58
FEDRawDataCollection
Definition: FEDRawDataCollection.h:18
evf::EvFDaqDirector::getLumisectionToStart
unsigned int getLumisectionToStart() const
Definition: EvFDaqDirector.cc:1890
edm
HLT enums.
Definition: AlignableModifier.h:19
FedRawDataInputSource::numConcurrentReads_
unsigned int numConcurrentReads_
Definition: FedRawDataInputSource.h:93
evf::FastMonState::inNoRequest
Definition: FastMonitoringService.h:98
evf::FastMonState::inSupNewFileWaitChunk
Definition: FastMonitoringService.h:116
evf::evtn::gtpe_board_sense
bool gtpe_board_sense(const unsigned char *p)
Definition: GlobalEventNumber.cc:11
FedRawDataInputSource::nStreams_
unsigned int nStreams_
Definition: FedRawDataInputSource.h:169
pos
Definition: PixelAliasList.h:18
FedRawDataInputSource::getNextEvent
evf::EvFDaqDirector::FileStatus getNextEvent()
Definition: FedRawDataInputSource.cc:352
FedRawDataInputSource::runNumber_
edm::RunNumber_t runNumber_
Definition: FedRawDataInputSource.h:111
FedRawDataInputSource::currentLumiSection_
unsigned int currentLumiSection_
Definition: FedRawDataInputSource.h:121
evf::FastMonState::inInit
Definition: FastMonitoringService.h:85
edm::ParameterSetDescription
Definition: ParameterSetDescription.h:52
l1tstage2_dqm_sourceclient-live_cfg.rawData
rawData
Definition: l1tstage2_dqm_sourceclient-live_cfg.py:163
FedRawDataInputSource::workerPool_
tbb::concurrent_queue< unsigned int > workerPool_
Definition: FedRawDataInputSource.h:148
FedRawDataInputSource::sourceEventsReport_
std::map< unsigned int, unsigned int > sourceEventsReport_
Definition: FedRawDataInputSource.h:183
DeadROCCounter.getRunNumber
def getRunNumber(filename)
Definition: DeadROCCounter.py:7
if
if(0==first)
Definition: CAHitNtupletGeneratorKernelsImpl.h:58
FedRawDataInputSource::readNextChunkIntoBuffer
void readNextChunkIntoBuffer(InputFile *file)
Definition: FedRawDataInputSource.cc:1436
edm::InputSourceDescription
Definition: InputSourceDescription.h:21
cms::cuda::assert
assert(be >=bs)
protons_cff.time
time
Definition: protons_cff.py:35
FedRawDataInputSource::fms_
evf::FastMonitoringService * fms_
Definition: FedRawDataInputSource.h:83
edm::second
U second(std::pair< T, U > const &p)
Definition: ParameterSet.cc:222
edm::Timestamp::beginOfTime
static Timestamp beginOfTime()
Definition: Timestamp.h:84
InputFile::moveToPreviousChunk
void moveToPreviousChunk(const size_t size, const size_t offset)
Definition: FedRawDataInputSource.cc:1399
FedRawDataInputSource::cvReader_
std::vector< std::condition_variable * > cvReader_
Definition: FedRawDataInputSource.h:155
FedRawDataInputSource::eventChunkSize_
unsigned int eventChunkSize_
Definition: FedRawDataInputSource.h:88
evf::evtn::get
unsigned int get(const unsigned char *, bool)
Definition: GlobalEventNumber.cc:77
FEDNumbering::MAXFEDID
Definition: FEDNumbering.h:26
FedRawDataInputSource::cvWakeup_
std::condition_variable cvWakeup_
Definition: FedRawDataInputSource.h:174
evf::EvFDaqDirector::sameFile
Definition: EvFDaqDirector.h:64
evf::EvFDaqDirector::runAbort
Definition: EvFDaqDirector.h:64
edm::LogInfo
Log< level::Info, false > LogInfo
Definition: MessageLogger.h:125
FedRawDataInputSource::freeChunks_
tbb::concurrent_queue< InputChunk * > freeChunks_
Definition: FedRawDataInputSource.h:151
MillePedeFileConverter_cfg.fileName
fileName
Definition: MillePedeFileConverter_cfg.py:32
evf::evtn::getgpslow
unsigned int getgpslow(const unsigned char *)
Definition: GlobalEventNumber.cc:92
patZpeak.events
events
Definition: patZpeak.py:20
newFWLiteAna.found
found
Definition: newFWLiteAna.py:118
edm::LogWarning
Log< level::Warning, false > LogWarning
Definition: MessageLogger.h:122
FRDHeaderVersionSize
constexpr std::array< uint32, FRDHeaderMaxVersion+1 > FRDHeaderVersionSize
Definition: FRDEventMessage.h:129
evf::FastMonState::inSupWaitFreeThread
Definition: FastMonitoringService.h:106
evf::FastMonitoringService::setInputSource
void setInputSource(FedRawDataInputSource *inputSource)
Definition: FastMonitoringService.h:217
FedRawDataInputSource::bufferInputRead_
uint32_t bufferInputRead_
Definition: FedRawDataInputSource.h:179
edm::InputSource::processHistoryRegistryForUpdate
ProcessHistoryRegistry & processHistoryRegistryForUpdate()
Definition: InputSource.h:331
edm::InputSource::state
ItemType state() const
Definition: InputSource.h:332
FEDRawData::data
const unsigned char * data() const
Return a const pointer to the beginning of the data buffer.
Definition: FEDRawData.cc:24
FedRawDataInputSource::getEventReport
std::pair< bool, unsigned int > getEventReport(unsigned int lumi, bool erase)
Definition: FedRawDataInputSource.cc:1516
InputFile::bufferPosition_
uint32_t bufferPosition_
Definition: FedRawDataInputSource.h:226
InputFile
Definition: FedRawDataInputSource.h:211
InputChunk::offset_
unsigned int offset_
Definition: FedRawDataInputSource.h:193
edm::Wrapper
Definition: Product.h:10
FedRawDataInputSource::workerJob_
std::vector< ReaderInfo > workerJob_
Definition: FedRawDataInputSource.h:149
FEDRawData
Definition: FEDRawData.h:19
InputFile::chunkPosition_
uint32_t chunkPosition_
Definition: FedRawDataInputSource.h:227
Utilities.operator
operator
Definition: Utilities.py:24
FedRawDataInputSource::readBlocks_
unsigned int readBlocks_
Definition: FedRawDataInputSource.h:90
AuxiliaryMakers.h
edm::LuminosityBlockAuxiliary
Definition: LuminosityBlockAuxiliary.h:15
FEDNumbering::MINTriggerGTPFEDID
Definition: FEDNumbering.h:61
evf::EvFDaqDirector::updateFuLock
FileStatus updateFuLock(unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint16_t &rawHeaderSize, uint64_t &lockWaitTime, bool &setExceptionState)
Definition: EvFDaqDirector.cc:503
FedRawDataInputSource::streamFileTracker_
std::vector< int > streamFileTracker_
Definition: FedRawDataInputSource.h:168
InputChunk::usedSize_
uint32_t usedSize_
Definition: FedRawDataInputSource.h:191
InputChunk::buf_
unsigned char * buf_
Definition: FedRawDataInputSource.h:188
InputChunk::reset
void reset(unsigned int newOffset, unsigned int toRead, unsigned int fileIndex)
Definition: FedRawDataInputSource.h:201
FedRawDataInputSource::eventRunNumber_
uint32_t eventRunNumber_
Definition: FedRawDataInputSource.h:122
evf::EvFDaqDirector::getEoLSFilePathOnFU
std::string getEoLSFilePathOnFU(const unsigned int ls) const
Definition: EvFDaqDirector.cc:480
evf::FastMonState::inReadCleanup
Definition: FastMonitoringService.h:97
FedRawDataInputSource::testTCDSFEDRange_
const std::vector< unsigned int > testTCDSFEDRange_
Definition: FedRawDataInputSource.h:101
dqmdumpme.last
last
Definition: dqmdumpme.py:56
FedRawDataInputSource::InputFile
friend struct InputFile
Definition: FedRawDataInputSource.h:41
edm::EventPrincipal
Definition: EventPrincipal.h:48
edm::ConfigurationDescriptions::add
void add(std::string const &label, ParameterSetDescription const &psetDescription)
Definition: ConfigurationDescriptions.cc:57
evf::EvFDaqDirector::isSingleStreamThread
bool isSingleStreamThread()
Definition: EvFDaqDirector.h:122
FEDNumbering::MAXTCDSuTCAFEDID
Definition: FEDNumbering.h:102
evf::FastMonState::inWaitInput
Definition: FastMonitoringService.h:86
evf::FastMonState::inNewLumi
Definition: FastMonitoringService.h:87
FRDHeaderMaxVersion
constexpr size_t FRDHeaderMaxVersion
Definition: FRDEventMessage.h:128
evf::EvFDaqDirector::parseFRDFileHeader
static int parseFRDFileHeader(std::string const &rawSourcePath, int &rawFd, uint16_t &rawHeaderSize, uint32_t &lsFromHeader, int32_t &eventsFromHeader, int64_t &fileSizeFromHeader, bool requireHeader, bool retry, bool closeFile)
Definition: EvFDaqDirector.cc:965
TCDSRaw.h
FedRawDataInputSource::filesToDelete_
std::list< std::pair< int, std::unique_ptr< InputFile > > > filesToDelete_
Definition: FedRawDataInputSource.h:165
EventID.h
FedRawDataInputSource::getLSFromFilename_
const bool getLSFromFilename_
Definition: FedRawDataInputSource.h:97
FedRawDataInputSource::ReaderInfo
std::pair< InputFile *, InputChunk * > ReaderInfo
Definition: FedRawDataInputSource.h:138
InputChunk
Definition: FedRawDataInputSource.h:187
FedRawDataInputSource::FedRawDataInputSource
FedRawDataInputSource(edm::ParameterSet const &, edm::InputSourceDescription const &)
Definition: FedRawDataInputSource.cc:53
mps_fire.end
end
Definition: mps_fire.py:242
FedRawDataInputSource::fileListLoopMode_
const bool fileListLoopMode_
Definition: FedRawDataInputSource.h:108
edm::InputSource::luminosityBlockAuxiliary
std::shared_ptr< LuminosityBlockAuxiliary > luminosityBlockAuxiliary() const
Called by the framework to merge or insert lumi in principal cache.
Definition: InputSource.h:233
InputFile::currentChunk_
unsigned int currentChunk_
Definition: FedRawDataInputSource.h:228
FedRawDataInputSource::daqDirector_
evf::EvFDaqDirector * daqDirector_
Definition: FedRawDataInputSource.h:84
evf::FastMonState::inSupFileLimit
Definition: FastMonitoringService.h:103
tcds::Raw_v1
Definition: TCDSRaw.h:106
svgfig.load
def load(fileName)
Definition: svgfig.py:547
dt4ml_dqm_sourceclient-live_cfg.filePath
filePath
CUSTOMIZE FOR ML.
Definition: dt4ml_dqm_sourceclient-live_cfg.py:45
evf::FastMonState::inChecksumEvent
Definition: FastMonitoringService.h:94
evf::EvFDaqDirector::noFile
Definition: EvFDaqDirector.h:64
FEDNumbering::MINTCDSuTCAFEDID
Definition: FEDNumbering.h:101
FedRawDataInputSource::fileQueue_
tbb::concurrent_queue< std::unique_ptr< InputFile > > fileQueue_
Definition: FedRawDataInputSource.h:152
evf::EvFDaqDirector::useFileBroker
bool useFileBroker() const
Definition: EvFDaqDirector.h:79
InputChunk::readComplete_
std::atomic< bool > readComplete_
Definition: FedRawDataInputSource.h:195
b
double b
Definition: hdecay.h:118
edm::EventPrincipal::put
void put(BranchDescription const &bd, std::unique_ptr< WrapperBase > edp, ProductProvenance const &productProvenance) const
Definition: EventPrincipal.cc:185
FedRawDataInputSource::reportEventsThisLumiInSource
void reportEventsThisLumiInSource(unsigned int lumi, unsigned int events)
Definition: FedRawDataInputSource.cc:1507
evf::FastMonState::inSupBusy
Definition: FastMonitoringService.h:108
FedRawDataInputSource::processHistoryID_
edm::ProcessHistoryID processHistoryID_
Definition: FedRawDataInputSource.h:119
InputSourceDescription.h
first
auto first
Definition: CAHitNtupletGeneratorKernelsImpl.h:125
InputFile::~InputFile
~InputFile()
Definition: FedRawDataInputSource.cc:1413
edm::EventAuxiliary
Definition: EventAuxiliary.h:14
edm::ConfigurationDescriptions
Definition: ConfigurationDescriptions.h:28
FedRawDataInputSource::startedSupervisorThread_
bool startedSupervisorThread_
Definition: FedRawDataInputSource.h:144
FedRawDataInputSource::exceptionState
bool exceptionState()
Definition: FedRawDataInputSource.h:68
FedRawDataInputSource::fillDescriptions
static void fillDescriptions(edm::ConfigurationDescriptions &descriptions)
Definition: FedRawDataInputSource.cc:206
FedRawDataInputSource::daqProvenanceHelper_
const edm::DaqProvenanceHelper daqProvenanceHelper_
Definition: FedRawDataInputSource.h:114
geometryDiff.file
file
Definition: geometryDiff.py:13
crc32c
uint32_t crc32c(uint32_t crc, const unsigned char *buf, size_t len)
Definition: crc32c.cc:340
edm::InputSource::runAuxiliary
std::shared_ptr< RunAuxiliary > runAuxiliary() const
Called by the framework to merge or insert run in principal cache.
Definition: InputSource.h:230
FedRawDataInputSource::initFileList
long initFileList()
Definition: FedRawDataInputSource.cc:1528
FedRawDataInputSource::L1EventID_
uint32_t L1EventID_
Definition: FedRawDataInputSource.h:124
UnixSignalHandlers.h
cppFunctionSkipper.exception
exception
Definition: cppFunctionSkipper.py:10
funct::true
true
Definition: Factorize.h:173
evf::FastMonState::inSupNewFile
Definition: FastMonitoringService.h:112
InputChunk::fileIndex_
unsigned int fileIndex_
Definition: FedRawDataInputSource.h:194
reader.h
LogDebug
#define LogDebug(id)
Definition: MessageLogger.h:233
edm::InputSource::setRunAuxiliary
void setRunAuxiliary(RunAuxiliary *rp)
Definition: InputSource.h:333
edm::ParameterSet
Definition: ParameterSet.h:47
FedRawDataInputSource::currentFileIndex_
int currentFileIndex_
Definition: FedRawDataInputSource.h:164
FEDTrailer
Definition: FEDTrailer.h:14
a
double a
Definition: hdecay.h:119
evf::FastMonState::inSupLockPolling
Definition: FastMonitoringService.h:109
Timestamp.h
FedRawDataInputSource::alwaysStartFromFirstLS_
const bool alwaysStartFromFirstLS_
Definition: FedRawDataInputSource.h:98
FFFNamingSchema.h
Event.h
edm::InputSource::productRegistryUpdate
ProductRegistry & productRegistryUpdate()
Definition: InputSource.h:330
FedRawDataInputSource::eventChunkBlock_
unsigned int eventChunkBlock_
Definition: FedRawDataInputSource.h:89
FEDHeader::triggerType
uint8_t triggerType() const
Event Trigger type identifier.
Definition: FEDHeader.cc:13
CommonMethods.lock
def lock()
Definition: CommonMethods.py:81
edm::DaqProvenanceHelper::daqInit
ProcessHistoryID daqInit(ProductRegistry &productRegistry, ProcessHistoryRegistry &processHistoryRegistry) const
Definition: DaqProvenanceHelper.cc:83
evf::EvFDaqDirector::getNextFromFileBroker
FileStatus getNextFromFileBroker(const unsigned int currentLumiSection, unsigned int &ls, std::string &nextFile, int &rawFd, uint16_t &rawHeaderSize, int32_t &serverEventsInNewFile_, int64_t &fileSize, uint64_t &thisLockWaitTimeUs)
Definition: EvFDaqDirector.cc:1717
FedRawDataInputSource::verifyChecksum_
const bool verifyChecksum_
Definition: FedRawDataInputSource.h:99
mps_check.timeout
int timeout
Definition: mps_check.py:53
evf::FastMonState::inReadEvent
Definition: FastMonitoringService.h:96
SiStripCommissioningSource_FromEDM_cfg.EvFDaqDirector
EvFDaqDirector
Definition: SiStripCommissioningSource_FromEDM_cfg.py:57
evf::EvFDaqDirector::unlockFULocal
void unlockFULocal()
Definition: EvFDaqDirector.cc:918
jetUpdater_cfi.sort
sort
Definition: jetUpdater_cfi.py:29
edm::shutdown_flag
volatile std::atomic< bool > shutdown_flag
Definition: UnixSignalHandlers.cc:22
FedRawDataInputSource::checkNext
Next checkNext() override
Definition: FedRawDataInputSource.cc:231
FedRawDataInputSource::fillFEDRawDataCollection
edm::Timestamp fillFEDRawDataCollection(FEDRawDataCollection &rawData, bool &tcdsInRange)
Definition: FedRawDataInputSource.cc:689
printConversionInfo.aux
aux
Definition: printConversionInfo.py:19
edm::Service
Definition: Service.h:30
createfilelist.int
int
Definition: createfilelist.py:10
FedRawDataInputSource::currentFile_
std::unique_ptr< InputFile > currentFile_
Definition: FedRawDataInputSource.h:141
FedRawDataInputSource::setExceptionState_
bool setExceptionState_
Definition: FedRawDataInputSource.h:160
FastMonitoringService.h
edm::InputSource::luminosityBlock
LuminosityBlockNumber_t luminosityBlock() const
Accessor for current luminosity block number.
Definition: InputSource.cc:462
FedRawDataInputSource::eventID_
edm::EventID eventID_
Definition: FedRawDataInputSource.h:118
edm::RawInputSource::Next::kStop
FEDRawDataCollection.h
evf::FastMonState::inSupWaitFreeChunk
Definition: FastMonitoringService.h:104
FedRawDataInputSource::mReader_
std::mutex mReader_
Definition: FedRawDataInputSource.h:154
trackerHitRTTI::vector
Definition: trackerHitRTTI.h:21
FedRawDataInputSource::eventsThisLumi_
unsigned int eventsThisLumi_
Definition: FedRawDataInputSource.h:126
crc32c.h
FedRawDataInputSource::~FedRawDataInputSource
~FedRawDataInputSource() override
Definition: FedRawDataInputSource.cc:175
FEDHeader::length
static const uint32_t length
Definition: FEDHeader.h:54
evf::EvFDaqDirector::getEoRFilePathOnFU
std::string getEoRFilePathOnFU() const
Definition: EvFDaqDirector.cc:490
edm::EventPrincipal::streamID
StreamID streamID() const
Definition: EventPrincipal.h:110
l1tstage2_dqm_sourceclient-live_cfg.fedId
fedId
Definition: l1tstage2_dqm_sourceclient-live_cfg.py:89
evf::EvFDaqDirector::newLumi
Definition: EvFDaqDirector.h:64
edm::LogError
Log< level::Error, false > LogError
Definition: MessageLogger.h:123
FedRawDataInputSource::getFile
evf::EvFDaqDirector::FileStatus getFile(unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint64_t &lockWaitTime)
Definition: FedRawDataInputSource.cc:1563
InputFile::rawFd_
int rawFd_
Definition: FedRawDataInputSource.h:217
FedRawDataInputSource::startupLock_
std::mutex startupLock_
Definition: FedRawDataInputSource.h:161
visDQMUpload.buf
buf
Definition: visDQMUpload.py:160
AlCaHLTBitMon_QueryRunRegistry.string
string string
Definition: AlCaHLTBitMon_QueryRunRegistry.py:256
tcds
Definition: TCDSRaw.h:16
FedRawDataInputSource.h
evf::FastMonitoringService::reportLockWait
void reportLockWait(unsigned int ls, double waitTime, unsigned int lockCount)
Definition: FastMonitoringService.cc:718
evf::evtn::evm_board_sense
bool evm_board_sense(const unsigned char *p, size_t size)
Definition: GlobalEventNumber.cc:15
FedRawDataInputSource::thread_quit_signal
std::vector< unsigned int > thread_quit_signal
Definition: FedRawDataInputSource.h:159
InputSourceMacros.h
InputFile::advance
bool advance(unsigned char *&dataPosition, const size_t size)
Definition: FedRawDataInputSource.cc:1359
evf::FastMonState::inSupNoFile
Definition: FastMonitoringService.h:111
FedRawDataInputSource::useFileBroker_
bool useFileBroker_
Definition: FedRawDataInputSource.h:103
edm::InputSource::setLuminosityBlockAuxiliary
void setLuminosityBlockAuxiliary(LuminosityBlockAuxiliary *lbp)
Definition: InputSource.h:337
evf::FastMonitoringService::setInState
void setInState(FastMonState::InputState inputState)
Definition: FastMonitoringService.h:218
DataPointDefinition.h
evf::FastMonState
Definition: FastMonitoringService.h:51
edm::RawInputSource::Next
Next
Definition: RawInputSource.h:24
edm::Timestamp::invalidTimestamp
static Timestamp invalidTimestamp()
Definition: Timestamp.h:82
edm::EventAuxiliary::PhysicsTrigger
Definition: EventAuxiliary.h:20
FedRawDataInputSource::readSupervisor
void readSupervisor()
Definition: FedRawDataInputSource.cc:752
submitPVResolutionJobs.desc
string desc
Definition: submitPVResolutionJobs.py:251
eostools.move
def move(src, dest)
Definition: eostools.py:511
std
Definition: JetResolutionObject.h:76
evf::EvFDaqDirector::createProcessingNotificationMaybe
void createProcessingNotificationMaybe() const
Definition: EvFDaqDirector.cc:2045
crc32c_hw_test
bool crc32c_hw_test()
Definition: crc32c.cc:354
dqmiodatasetharvest.inf
inf
Definition: dqmiodatasetharvest.py:38
init
Definition: init.py:1
RunInfoPI::state
state
Definition: RunInfoPayloadInspectoHelper.h:16
InputFile::chunks_
tbb::concurrent_vector< InputChunk * > chunks_
Definition: FedRawDataInputSource.h:224
edm::RawInputSource::makeEvent
void makeEvent(EventPrincipal &eventPrincipal, EventAuxiliary const &eventAuxiliary)
Definition: RawInputSource.cc:51
edm::DaqProvenanceHelper::branchDescription
BranchDescription const & branchDescription() const
Definition: DaqProvenanceHelper.h:46
FedRawDataInputSource::event_
std::unique_ptr< FRDEventMsgView > event_
Definition: FedRawDataInputSource.h:116
edm::InputSource::setNewRun
void setNewRun()
Definition: InputSource.h:355
FEDTrailer::length
static const uint32_t length
Definition: FEDTrailer.h:57
FedRawDataInputSource::nextEvent
evf::EvFDaqDirector::FileStatus nextEvent()
Definition: FedRawDataInputSource.cc:343
FedRawDataInputSource::detectedFRDversion_
uint16_t detectedFRDversion_
Definition: FedRawDataInputSource.h:140
relativeConstraints.empty
bool empty
Definition: relativeConstraints.py:46
GlobalEventNumber.h
EventAuxiliary.h
FedRawDataInputSource::tid_active_
std::vector< unsigned int > tid_active_
Definition: FedRawDataInputSource.h:156
Exception
Definition: hltDiff.cc:245
FedRawDataInputSource::checkEvery_
unsigned int checkEvery_
Definition: FedRawDataInputSource.h:170
evf::FastMonState::inProcessingFile
Definition: FastMonitoringService.h:91
FedRawDataInputSource::setMonStateSup
void setMonStateSup(evf::FastMonState::InputState state)
Definition: FedRawDataInputSource.cc:1354
FEDRawData::resize
void resize(size_t newsize)
Definition: FEDRawData.cc:28
evf::EvFDaqDirector::setFMS
void setFMS(evf::FastMonitoringService *fms)
Definition: EvFDaqDirector.h:121
MatrixUtil.remove
def remove(d, key, TELL=False)
Definition: MatrixUtil.py:219
FedRawDataInputSource::fileDescriptor_
int fileDescriptor_
Definition: FedRawDataInputSource.h:178
evf
Definition: fillJson.h:27
edm::InputSource::setEventCached
void setEventCached()
Called by the framework to merge or ached() const {return eventCached_;}.
Definition: InputSource.h:362
evf::FastMonitoringService::startedLookingForFile
void startedLookingForFile()
Definition: FastMonitoringService.cc:682
evf::FastMonState::inCachedEvent
Definition: FastMonitoringService.h:95
evf::FastMonState::inSupNewFileWaitThreadCopying
Definition: FastMonitoringService.h:113
evf::FastMonitoringService::setExceptionDetected
void setExceptionDetected(unsigned int ls)
Definition: FastMonitoringService.cc:417
evf::EvFDaqDirector::grabNextJsonFileAndUnlock
int grabNextJsonFileAndUnlock(std::filesystem::path const &jsonSourcePath)
Definition: EvFDaqDirector.cc:1362
edm::InputSource::resetLuminosityBlockAuxiliary
void resetLuminosityBlockAuxiliary(bool isNewLumi=true) const
Definition: InputSource.h:345
evf::FastMonState::inRunEnd
Definition: FastMonitoringService.h:90
FEDHeader::sourceID
uint16_t sourceID() const
Identifier of the FED.
Definition: FEDHeader.cc:19
FedRawDataInputSource::startupCv_
std::condition_variable startupCv_
Definition: FedRawDataInputSource.h:162
FedRawDataInputSource::tcds_pointer_
unsigned char * tcds_pointer_
Definition: FedRawDataInputSource.h:125
cond::uint64_t
unsigned long long uint64_t
Definition: Time.h:13
funct::pow
Power< A, B >::type pow(const A &a, const B &b)
Definition: Power.h:29
FedRawDataInputSource::fileListMode_
const bool fileListMode_
Definition: FedRawDataInputSource.h:106
cms::Exception
Definition: Exception.h:70
castor_dqm_sourceclient_file_cfg.path
path
Definition: castor_dqm_sourceclient_file_cfg.py:37
FedRawDataInputSource::InputChunk
friend struct InputChunk
Definition: FedRawDataInputSource.h:42
evf::FastMonState::inSupWaitFreeChunkCopying
Definition: FastMonitoringService.h:105
FedRawDataInputSource::readSupervisorThread_
std::unique_ptr< std::thread > readSupervisorThread_
Definition: FedRawDataInputSource.h:145
command_line.start
start
Definition: command_line.py:167
ParameterSet.h
edm::InputSource::processGUID
std::string const & processGUID() const
Accessor for global process identifier.
Definition: InputSource.h:195
dqmiolumiharvest.j
j
Definition: dqmiolumiharvest.py:66
runEdmFileComparison.skipped
skipped
Definition: runEdmFileComparison.py:224
edm_modernize_messagelogger.stat
stat
Definition: edm_modernize_messagelogger.py:27
edm::RawInputSource::Next::kEvent
evf::evtn::gtpe_get
unsigned int gtpe_get(const unsigned char *)
Definition: GlobalEventNumber.cc:83
event
Definition: event.py:1
edm::EventID
Definition: EventID.h:31
evf::FastMonitoringService::stoppedLookingForFile
void stoppedLookingForFile(unsigned int lumi)
Definition: FastMonitoringService.cc:690
hltrates_dqm_sourceclient-live_cfg.offset
offset
Definition: hltrates_dqm_sourceclient-live_cfg.py:82
FEDHeader
Definition: FEDHeader.h:14
InputFile::waitForChunk
bool waitForChunk(unsigned int chunkid)
Definition: FedRawDataInputSource.h:258
FedRawDataInputSource::MAXTCDSuTCAFEDID_
uint16_t MAXTCDSuTCAFEDID_
Definition: FedRawDataInputSource.h:130
FEDHeader.h
lumi
Definition: LumiSectionData.h:20
InputFile::rewindChunk
void rewindChunk(const size_t size)
Definition: FedRawDataInputSource.cc:1408
edm::Log
Definition: MessageLogger.h:70
evf::EvFDaqDirector::FileStatus
FileStatus
Definition: EvFDaqDirector.h:64
evf::FastMonitoringService::setInStateSup
void setInStateSup(FastMonState::InputState inputState)
Definition: FastMonitoringService.h:219
evf::EvFDaqDirector::lockFULocal2
void lockFULocal2()
Definition: EvFDaqDirector.cc:923
FedRawDataInputSource::monlock_
std::mutex monlock_
Definition: FedRawDataInputSource.h:184
InputFile::deleteFile_
bool deleteFile_
Definition: FedRawDataInputSource.h:216
FedRawDataInputSource::read
void read(edm::EventPrincipal &eventPrincipal) override
Definition: FedRawDataInputSource.cc:613
sistrip::runNumber_
static const char runNumber_[]
Definition: ConstantsForDqm.h:33
FedRawDataInputSource::numBuffers_
unsigned int numBuffers_
Definition: FedRawDataInputSource.h:91
evf::evtn::getgpshigh
unsigned int getgpshigh(const unsigned char *)
Definition: GlobalEventNumber.cc:95
FedRawDataInputSource::maxBufferedFiles_
unsigned int maxBufferedFiles_
Definition: FedRawDataInputSource.h:92
edm::TimeValue_t
unsigned long long TimeValue_t
Definition: Timestamp.h:28
muonDTDigis_cfi.pset
pset
Definition: muonDTDigis_cfi.py:27
FedRawDataInputSource::singleBufferMode_
bool singleBufferMode_
Definition: FedRawDataInputSource.h:177
FedRawDataInputSource::readWorker
void readWorker(unsigned int tid)
Definition: FedRawDataInputSource.cc:1202
FedRawDataInputSource::loopModeIterationInc_
unsigned int loopModeIterationInc_
Definition: FedRawDataInputSource.h:109
evf::EvFDaqDirector::newFile
Definition: EvFDaqDirector.h:64
edm::RunAuxiliary
Definition: RunAuxiliary.h:15
findQualityFiles.size
size
Write out results.
Definition: findQualityFiles.py:443
edm::InputSource::run
RunNumber_t run() const
Accessor for current run number.
Definition: InputSource.cc:457
evf::EvFDaqDirector::grabNextJsonFromRaw
int grabNextJsonFromRaw(std::string const &rawSourcePath, int &rawFd, uint16_t &rawHeaderSize, int64_t &fileSizeFromHeader, bool &fileFound, uint32_t serverLS, bool closeFile)
Definition: EvFDaqDirector.cc:1100
FEDTrailer::fragmentLength
uint32_t fragmentLength() const
The length of the event fragment counted in 64-bit words including header and trailer.
Definition: FEDTrailer.cc:13
FedRawDataInputSource::GTPEventID_
uint32_t GTPEventID_
Definition: FedRawDataInputSource.h:123
evf::EvFDaqDirector::setDeleteTracking
void setDeleteTracking(std::mutex *fileDeleteLock, std::list< std::pair< int, std::unique_ptr< InputFile >>> *filesToDelete)
Definition: EvFDaqDirector.h:176
edm::DaqProvenanceHelper::dummyProvenance
ProductProvenance const & dummyProvenance() const
Definition: DaqProvenanceHelper.h:48
FEDTrailer.h
evf::FastMonState::InputState
InputState
Definition: FastMonitoringService.h:83
edm::Timestamp
Definition: Timestamp.h:30