CMS 3D CMS Logo

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 using namespace evf::FastMonState;
50 
52  : edm::RawInputSource(pset, desc),
53  defPath_(pset.getUntrackedParameter<std::string>("buDefPath", "")),
54  eventChunkSize_(pset.getUntrackedParameter<unsigned int>("eventChunkSize", 32) * 1048576),
55  eventChunkBlock_(pset.getUntrackedParameter<unsigned int>("eventChunkBlock", 32) * 1048576),
56  numBuffers_(pset.getUntrackedParameter<unsigned int>("numBuffers", 2)),
57  maxBufferedFiles_(pset.getUntrackedParameter<unsigned int>("maxBufferedFiles", 2)),
58  getLSFromFilename_(pset.getUntrackedParameter<bool>("getLSFromFilename", true)),
59  alwaysStartFromFirstLS_(pset.getUntrackedParameter<bool>("alwaysStartFromFirstLS", false)),
60  verifyChecksum_(pset.getUntrackedParameter<bool>("verifyChecksum", true)),
61  useL1EventID_(pset.getUntrackedParameter<bool>("useL1EventID", false)),
62  testTCDSFEDRange_(
63  pset.getUntrackedParameter<std::vector<unsigned int>>("testTCDSFEDRange", std::vector<unsigned int>())),
64  fileNames_(pset.getUntrackedParameter<std::vector<std::string>>("fileNames", std::vector<std::string>())),
65  fileListMode_(pset.getUntrackedParameter<bool>("fileListMode", false)),
66  fileListLoopMode_(pset.getUntrackedParameter<bool>("fileListLoopMode", false)),
67  runNumber_(edm::Service<evf::EvFDaqDirector>()->getRunNumber()),
68  daqProvenanceHelper_(edm::TypeID(typeid(FEDRawDataCollection))),
69  eventID_(),
70  processHistoryID_(),
71  currentLumiSection_(0),
72  tcds_pointer_(nullptr),
73  eventsThisLumi_(0) {
74  char thishost[256];
75  gethostname(thishost, 255);
76  edm::LogInfo("FedRawDataInputSource") << "Construction. read-ahead chunk size -: " << std::endl
77  << (eventChunkSize_ / 1048576) << " MB on host " << thishost;
78 
79  if (!testTCDSFEDRange_.empty()) {
80  if (testTCDSFEDRange_.size() != 2) {
81  throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection")
82  << "Invalid TCDS Test FED range parameter";
83  }
86  }
87 
88  long autoRunNumber = -1;
89  if (fileListMode_) {
90  autoRunNumber = initFileList();
91  if (!fileListLoopMode_) {
92  if (autoRunNumber < 0)
93  throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource") << "Run number not found from filename";
94  //override run number
95  runNumber_ = (edm::RunNumber_t)autoRunNumber;
96  edm::Service<evf::EvFDaqDirector>()->overrideRunNumber((unsigned int)autoRunNumber);
97  }
98  }
99 
101  setNewRun();
102  //todo:autodetect from file name (assert if names differ)
104 
105  //make sure that chunk size is N * block size
110 
111  if (!numBuffers_)
112  throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource")
113  << "no reading enabled with numBuffers parameter 0";
114 
117  readingFilesCount_ = 0;
118 
119  if (!crc32c_hw_test())
120  edm::LogError("FedRawDataInputSource::FedRawDataInputSource") << "Intel crc32c checksum computation unavailable";
121 
122  //get handles to DaqDirector and FastMonitoringService because getting them isn't possible in readSupervisor thread
123  if (fileListMode_) {
124  try {
126  } catch (cms::Exception const&) {
127  edm::LogInfo("FedRawDataInputSource") << "No FastMonitoringService found in the configuration";
128  }
129  } else {
131  if (!fms_) {
132  throw cms::Exception("FedRawDataInputSource") << "FastMonitoringService not found";
133  }
134  }
135 
137  if (!daqDirector_)
138  cms::Exception("FedRawDataInputSource") << "EvFDaqDirector not found";
139 
141  if (useFileBroker_)
142  edm::LogInfo("FedRawDataInputSource") << "EvFDaqDirector/Source configured to use file service";
143  //set DaqDirector to delete files in preGlobalEndLumi callback
145  if (fms_) {
147  fms_->setInputSource(this);
150  }
151  //should delete chunks when run stops
152  for (unsigned int i = 0; i < numBuffers_; i++) {
154  }
155 
156  quit_threads_ = false;
157 
158  //prepare data shared by threads
159  for (unsigned int i = 0; i < numConcurrentReads_; i++) {
160  thread_quit_signal.push_back(false);
161  workerJob_.push_back(ReaderInfo(nullptr, nullptr));
162  cvReader_.push_back(std::make_unique<std::condition_variable>());
163  tid_active_.push_back(0);
164  }
165 
166  //start threads
167  for (unsigned int i = 0; i < numConcurrentReads_; i++) {
168  //wait for each thread to complete initialization
169  std::unique_lock<std::mutex> lk(startupLock_);
170  workerThreads_.push_back(new std::thread(&FedRawDataInputSource::readWorker, this, i));
171  startupCv_.wait(lk);
172  }
173 
174  runAuxiliary()->setProcessHistoryID(processHistoryID_);
175 }
176 
178  quit_threads_ = true;
179 
180  //delete any remaining open files
181  if (!fms_ || !fms_->exceptionDetected()) {
182  for (auto it = filesToDelete_.begin(); it != filesToDelete_.end(); it++)
183  it->second.reset();
184  } else {
185  //skip deleting files with exception
186  for (auto it = filesToDelete_.begin(); it != filesToDelete_.end(); it++) {
187  //it->second->unsetDeleteFile();
188  if (fms_->isExceptionOnData(it->second->lumi_))
189  it->second->unsetDeleteFile();
190  else
191  it->second.reset();
192  }
193  //disable deleting current file with exception
194  if (currentFile_.get())
195  if (fms_->isExceptionOnData(currentFile_->lumi_))
196  currentFile_->unsetDeleteFile();
197  }
198 
200  readSupervisorThread_->join();
201  } else {
202  //join aux threads in case the supervisor thread was not started
203  for (unsigned int i = 0; i < workerThreads_.size(); i++) {
204  std::unique_lock<std::mutex> lk(mReader_);
205  thread_quit_signal[i] = true;
206  cvReader_[i]->notify_one();
207  lk.unlock();
208  workerThreads_[i]->join();
209  delete workerThreads_[i];
210  }
211  }
212 }
213 
216  desc.setComment("File-based Filter Farm input source for reading raw data from BU ramdisk");
217  desc.addUntracked<unsigned int>("eventChunkSize", 32)->setComment("Input buffer (chunk) size");
218  desc.addUntracked<unsigned int>("eventChunkBlock", 32)
219  ->setComment("Block size used in a single file read call (must be smaller or equal to buffer size)");
220  desc.addUntracked<unsigned int>("numBuffers", 2)->setComment("Number of buffers used for reading input");
221  desc.addUntracked<unsigned int>("maxBufferedFiles", 2)
222  ->setComment("Maximum number of simultaneously buffered raw files");
223  desc.addUntracked<unsigned int>("alwaysStartFromfirstLS", false)
224  ->setComment("Force source to start from LS 1 if server provides higher lumisection number");
225  desc.addUntracked<bool>("verifyChecksum", true)
226  ->setComment("Verify event CRC-32C checksum of FRDv5 and higher or Adler32 with v3 and v4");
227  desc.addUntracked<bool>("useL1EventID", false)
228  ->setComment("Use L1 event ID from FED header if true or from TCDS FED if false");
229  desc.addUntracked<std::vector<unsigned int>>("testTCDSFEDRange", std::vector<unsigned int>())
230  ->setComment("[min, max] range to search for TCDS FED ID in test setup");
231  desc.addUntracked<bool>("fileListMode", false)
232  ->setComment("Use fileNames parameter to directly specify raw files to open");
233  desc.addUntracked<std::vector<std::string>>("fileNames", std::vector<std::string>())
234  ->setComment("file list used when fileListMode is enabled");
235  desc.setAllowAnything();
236  descriptions.add("source", desc);
237 }
238 
241  //this thread opens new files and dispatches reading to worker readers
242  std::unique_lock<std::mutex> lk(startupLock_);
243  readSupervisorThread_ = std::make_unique<std::thread>(&FedRawDataInputSource::readSupervisor, this);
245  startupCv_.wait(lk);
246  }
247  //signal hltd to start event accounting
248  if (!currentLumiSection_)
251  switch (nextEvent()) {
253  //maybe create EoL file in working directory before ending run
254  struct stat buf;
255  if (!useFileBroker_ && currentLumiSection_ > 0) {
256  bool eolFound = (stat(daqDirector_->getEoLSFilePathOnBU(currentLumiSection_).c_str(), &buf) == 0);
257  if (eolFound) {
259  bool found = (stat(fuEoLS.c_str(), &buf) == 0);
260  if (!found) {
262  int eol_fd =
263  open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
264  close(eol_fd);
266  }
267  }
268  }
269  //also create EoR file in FU data directory
270  bool eorFound = (stat(daqDirector_->getEoRFilePathOnFU().c_str(), &buf) == 0);
271  if (!eorFound) {
272  int eor_fd = open(daqDirector_->getEoRFilePathOnFU().c_str(),
273  O_RDWR | O_CREAT,
274  S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
275  close(eor_fd);
276  }
278  eventsThisLumi_ = 0;
280  edm::LogInfo("FedRawDataInputSource") << "----------------RUN ENDED----------------";
281  return Next::kStop;
282  }
284  //this is not reachable
285  return Next::kEvent;
286  }
288  //std::cout << "--------------NEW LUMI---------------" << std::endl;
289  return Next::kEvent;
290  }
291  default: {
292  if (!getLSFromFilename_) {
293  //get new lumi from file header
294  if (event_->lumi() > currentLumiSection_) {
296  eventsThisLumi_ = 0;
298  }
299  }
302  else
303  eventRunNumber_ = event_->run();
304  L1EventID_ = event_->event();
305 
306  setEventCached();
307 
308  return Next::kEvent;
309  }
310  }
311 }
312 
313 void FedRawDataInputSource::maybeOpenNewLumiSection(const uint32_t lumiSection) {
314  if (!luminosityBlockAuxiliary() || luminosityBlockAuxiliary()->luminosityBlock() != lumiSection) {
315  if (!useFileBroker_) {
316  if (currentLumiSection_ > 0) {
318  struct stat buf;
319  bool found = (stat(fuEoLS.c_str(), &buf) == 0);
320  if (!found) {
322  int eol_fd =
323  open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
324  close(eol_fd);
325  daqDirector_->createBoLSFile(lumiSection, false);
327  }
328  } else
329  daqDirector_->createBoLSFile(lumiSection, true); //needed for initial lumisection
330  }
331 
332  currentLumiSection_ = lumiSection;
333 
335 
336  timeval tv;
337  gettimeofday(&tv, nullptr);
338  const edm::Timestamp lsopentime((unsigned long long)tv.tv_sec * 1000000 + (unsigned long long)tv.tv_usec);
339 
341  runAuxiliary()->run(), lumiSection, lsopentime, edm::Timestamp::invalidTimestamp());
342 
343  setLuminosityBlockAuxiliary(lumiBlockAuxiliary);
344  luminosityBlockAuxiliary()->setProcessHistoryID(processHistoryID_);
345 
346  edm::LogInfo("FedRawDataInputSource") << "New lumi section was opened. LUMI -: " << lumiSection;
347  }
348 }
349 
353  if (edm::shutdown_flag.load(std::memory_order_relaxed))
354  break;
355  }
356  return status;
357 }
358 
360  if (setExceptionState_)
361  threadError();
362  if (!currentFile_.get()) {
365  if (!fileQueue_.try_pop(currentFile_)) {
366  //sleep until wakeup (only in single-buffer mode) or timeout
367  std::unique_lock<std::mutex> lkw(mWakeup_);
368  if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout || !currentFile_.get())
370  }
371  status = currentFile_->status_;
374  currentFile_.reset();
375  return status;
376  } else if (status == evf::EvFDaqDirector::runAbort) {
377  throw cms::Exception("FedRawDataInputSource::getNextEvent")
378  << "Run has been aborted by the input source reader thread";
379  } else if (status == evf::EvFDaqDirector::newLumi) {
381  if (getLSFromFilename_) {
382  if (currentFile_->lumi_ > currentLumiSection_) {
384  eventsThisLumi_ = 0;
386  }
387  } else { //let this be picked up from next event
389  }
390  currentFile_.reset();
391  return status;
392  } else if (status == evf::EvFDaqDirector::newFile) {
394  } else
395  assert(false);
396  }
398 
399  //file is empty
400  if (!currentFile_->fileSize_) {
402  //try to open new lumi
403  assert(currentFile_->nChunks_ == 0);
404  if (getLSFromFilename_)
405  if (currentFile_->lumi_ > currentLumiSection_) {
407  eventsThisLumi_ = 0;
409  }
410  //immediately delete empty file
411  currentFile_.reset();
413  }
414 
415  //file is finished
416  if (currentFile_->bufferPosition_ == currentFile_->fileSize_) {
418  //release last chunk (it is never released elsewhere)
419  freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_]);
420  if (currentFile_->nEvents_ >= 0 && currentFile_->nEvents_ != int(currentFile_->nProcessed_)) {
421  throw cms::Exception("FedRawDataInputSource::getNextEvent")
422  << "Fully processed " << currentFile_->nProcessed_ << " from the file " << currentFile_->fileName_
423  << " but according to BU JSON there should be " << currentFile_->nEvents_ << " events";
424  }
425  //try to wake up supervisor thread which might be sleeping waiting for the free chunk
426  if (singleBufferMode_) {
427  std::unique_lock<std::mutex> lkw(mWakeup_);
428  cvWakeup_.notify_one();
429  }
430  bufferInputRead_ = 0;
432  //put the file in pending delete list;
433  std::unique_lock<std::mutex> lkw(fileDeleteLock_);
434  filesToDelete_.push_back(std::pair<int, std::unique_ptr<InputFile>>(currentFileIndex_, std::move(currentFile_)));
435  } else {
436  //in single-thread and stream jobs, events are already processed
437  currentFile_.reset();
438  }
440  }
441 
442  //handle RAW file header
443  if (currentFile_->bufferPosition_ == 0 && currentFile_->rawHeaderSize_ > 0) {
444  if (currentFile_->fileSize_ <= currentFile_->rawHeaderSize_) {
445  if (currentFile_->fileSize_ < currentFile_->rawHeaderSize_)
446  throw cms::Exception("FedRawDataInputSource::getNextEvent")
447  << "Premature end of input file while reading file header";
448 
449  edm::LogWarning("FedRawDataInputSource")
450  << "File with only raw header and no events received in LS " << currentFile_->lumi_;
451  if (getLSFromFilename_)
452  if (currentFile_->lumi_ > currentLumiSection_) {
454  eventsThisLumi_ = 0;
456  }
457  }
458 
459  //advance buffer position to skip file header (chunk will be acquired later)
460  currentFile_->chunkPosition_ += currentFile_->rawHeaderSize_;
461  currentFile_->bufferPosition_ += currentFile_->rawHeaderSize_;
462  }
463 
464  //file is too short
465  if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < FRDHeaderVersionSize[detectedFRDversion_]) {
466  throw cms::Exception("FedRawDataInputSource::getNextEvent")
467  << "Premature end of input file while reading event header";
468  }
469  if (singleBufferMode_) {
470  //should already be there
472  while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
473  usleep(10000);
474  if (currentFile_->parent_->exceptionState() || setExceptionState_)
475  currentFile_->parent_->threadError();
476  }
478 
479  unsigned char* dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
480 
481  //conditions when read amount is not sufficient for the header to fit
485 
486  if (detectedFRDversion_ == 0) {
487  detectedFRDversion_ = *((uint16_t*)dataPosition);
489  throw cms::Exception("FedRawDataInputSource::getNextEvent")
490  << "Unknown FRD version -: " << detectedFRDversion_;
492  }
493 
494  //recalculate chunk position
495  dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
497  throw cms::Exception("FedRawDataInputSource::getNextEvent")
498  << "Premature end of input file while reading event header";
499  }
500  }
501 
502  event_ = std::make_unique<FRDEventMsgView>(dataPosition);
503  if (event_->size() > eventChunkSize_) {
504  throw cms::Exception("FedRawDataInputSource::getNextEvent")
505  << " event id:" << event_->event() << " lumi:" << event_->lumi() << " run:" << event_->run()
506  << " of size:" << event_->size() << " bytes does not fit into a chunk of size:" << eventChunkSize_
507  << " bytes";
508  }
509 
510  const uint32_t msgSize = event_->size() - FRDHeaderVersionSize[detectedFRDversion_];
511 
512  if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize) {
513  throw cms::Exception("FedRawDataInputSource::getNextEvent")
514  << "Premature end of input file while reading event data";
515  }
516  if (eventChunkSize_ - currentFile_->chunkPosition_ < msgSize) {
518  //recalculate chunk position
519  dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
520  event_ = std::make_unique<FRDEventMsgView>(dataPosition);
521  }
522  currentFile_->bufferPosition_ += event_->size();
523  currentFile_->chunkPosition_ += event_->size();
524  //last chunk is released when this function is invoked next time
525 
526  }
527  //multibuffer mode:
528  else {
529  //wait for the current chunk to become added to the vector
531  while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
532  usleep(10000);
533  if (setExceptionState_)
534  threadError();
535  }
537 
538  //check if header is at the boundary of two chunks
539  chunkIsFree_ = false;
540  unsigned char* dataPosition;
541 
542  //read header, copy it to a single chunk if necessary
544  throw cms::Exception("FedRawDataInputSource::getNextEvent")
545  << "Premature end of input file (missing:"
547  << ") while reading event data for next event header";
548  bool chunkEnd = currentFile_->advance(dataPosition, FRDHeaderVersionSize[detectedFRDversion_]);
549 
550  event_ = std::make_unique<FRDEventMsgView>(dataPosition);
551  if (event_->size() > eventChunkSize_) {
552  throw cms::Exception("FedRawDataInputSource::getNextEvent")
553  << " event id:" << event_->event() << " lumi:" << event_->lumi() << " run:" << event_->run()
554  << " of size:" << event_->size() << " bytes does not fit into a chunk of size:" << eventChunkSize_
555  << " bytes";
556  }
557 
558  const uint32_t msgSize = event_->size() - FRDHeaderVersionSize[detectedFRDversion_];
559 
560  if (currentFile_->fileSizeLeft() < msgSize) {
561  throw cms::Exception("FedRawDataInputSource::getNextEvent")
562  << "Premature end of input file (missing:" << (msgSize - currentFile_->fileSizeLeft())
563  << ") while reading event data for event " << event_->event() << " lumi:" << event_->lumi();
564  }
565 
566  if (chunkEnd) {
567  //header was at the chunk boundary, we will have to move payload as well
568  currentFile_->moveToPreviousChunk(msgSize, FRDHeaderVersionSize[detectedFRDversion_]);
569  chunkIsFree_ = true;
570  } else {
571  //header was contiguous, but check if payload fits the chunk
572  if (eventChunkSize_ - currentFile_->chunkPosition_ < msgSize) {
573  //rewind to header start position
575  //copy event to a chunk start and move pointers
576 
578 
579  chunkEnd = currentFile_->advance(dataPosition, FRDHeaderVersionSize[detectedFRDversion_] + msgSize);
580 
582 
583  assert(chunkEnd);
584  chunkIsFree_ = true;
585  //header is moved
586  event_ = std::make_unique<FRDEventMsgView>(dataPosition);
587  } else {
588  //everything is in a single chunk, only move pointers forward
589  chunkEnd = currentFile_->advance(dataPosition, msgSize);
590  assert(!chunkEnd);
591  chunkIsFree_ = false;
592  }
593  }
594  //sanity-check check that the buffer position has not exceeded file size after preparing event
595  if (currentFile_->fileSize_ < currentFile_->bufferPosition_) {
596  throw cms::Exception("FedRawDataInputSource::getNextEvent")
597  << "Exceeded file size by " << currentFile_->bufferPosition_ - currentFile_->fileSize_
598  << " after reading last event declared size of " << event_->size() << " bytes";
599  }
600  } //end multibuffer mode
602 
603  if (verifyChecksum_ && event_->version() >= 5) {
604  uint32_t crc = 0;
605  crc = crc32c(crc, (const unsigned char*)event_->payload(), event_->eventSize());
606  if (crc != event_->crc32c()) {
607  if (fms_)
609  throw cms::Exception("FedRawDataInputSource::getNextEvent")
610  << "Found a wrong crc32c checksum: expected 0x" << std::hex << event_->crc32c() << " but calculated 0x"
611  << crc;
612  }
613  } else if (verifyChecksum_ && event_->version() >= 3) {
614  uint32_t adler = adler32(0L, Z_NULL, 0);
615  adler = adler32(adler, (Bytef*)event_->payload(), event_->eventSize());
616 
617  if (adler != event_->adler32()) {
618  if (fms_)
620  throw cms::Exception("FedRawDataInputSource::getNextEvent")
621  << "Found a wrong Adler32 checksum: expected 0x" << std::hex << event_->adler32() << " but calculated 0x"
622  << adler;
623  }
624  }
626 
627  currentFile_->nProcessed_++;
628 
630 }
631 
634  std::unique_ptr<FEDRawDataCollection> rawData(new FEDRawDataCollection);
635  bool tcdsInRange;
636  edm::Timestamp tstamp = fillFEDRawDataCollection(*rawData, tcdsInRange);
637 
638  if (useL1EventID_) {
641  aux.setProcessHistoryID(processHistoryID_);
642  makeEvent(eventPrincipal, aux);
643  } else if (tcds_pointer_ == nullptr) {
644  if (!GTPEventID_) {
645  throw cms::Exception("FedRawDataInputSource::read")
646  << "No TCDS or GTP FED in event with FEDHeader EID -: " << L1EventID_;
647  }
650  aux.setProcessHistoryID(processHistoryID_);
651  makeEvent(eventPrincipal, aux);
652  } else {
653  const FEDHeader fedHeader(tcds_pointer_);
654  tcds::Raw_v1 const* tcds = reinterpret_cast<tcds::Raw_v1 const*>(tcds_pointer_ + FEDHeader::length);
659  event_->isRealData(),
660  static_cast<edm::EventAuxiliary::ExperimentType>(fedHeader.triggerType()),
661  processGUID(),
663  !tcdsInRange);
664  aux.setProcessHistoryID(processHistoryID_);
665  makeEvent(eventPrincipal, aux);
666  }
667 
668  std::unique_ptr<edm::WrapperBase> edp(new edm::Wrapper<FEDRawDataCollection>(std::move(rawData)));
669 
671 
672  eventsThisLumi_++;
674 
675  //resize vector if needed
676  while (streamFileTracker_.size() <= eventPrincipal.streamID())
677  streamFileTracker_.push_back(-1);
678 
679  streamFileTracker_[eventPrincipal.streamID()] = currentFileIndex_;
680 
681  //this old file check runs no more often than every 10 events
682  if (!((currentFile_->nProcessed_ - 1) % (checkEvery_))) {
683  //delete files that are not in processing
684  std::unique_lock<std::mutex> lkw(fileDeleteLock_);
685  auto it = filesToDelete_.begin();
686  while (it != filesToDelete_.end()) {
687  bool fileIsBeingProcessed = false;
688  for (unsigned int i = 0; i < streamFileTracker_.size(); i++) {
689  if (it->first == streamFileTracker_.at(i)) {
690  fileIsBeingProcessed = true;
691  break;
692  }
693  }
694  if (!fileIsBeingProcessed && (!fms_ || !fms_->isExceptionOnData(it->second->lumi_))) {
695  std::string fileToDelete = it->second->fileName_;
696  it = filesToDelete_.erase(it);
697  } else
698  it++;
699  }
700  }
701  if (chunkIsFree_)
702  freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_ - 1]);
703  chunkIsFree_ = false;
705  return;
706 }
707 
710  timeval stv;
711  gettimeofday(&stv, nullptr);
712  time = stv.tv_sec;
713  time = (time << 32) + stv.tv_usec;
714  edm::Timestamp tstamp(time);
715 
716  uint32_t eventSize = event_->eventSize();
717  unsigned char* event = (unsigned char*)event_->payload();
718  GTPEventID_ = 0;
719  tcds_pointer_ = nullptr;
720  tcdsInRange = false;
721  uint16_t selectedTCDSFed = 0;
722  while (eventSize > 0) {
723  assert(eventSize >= FEDTrailer::length);
724  eventSize -= FEDTrailer::length;
725  const FEDTrailer fedTrailer(event + eventSize);
726  const uint32_t fedSize = fedTrailer.fragmentLength() << 3; //trailer length counts in 8 bytes
727  assert(eventSize >= fedSize - FEDHeader::length);
728  eventSize -= (fedSize - FEDHeader::length);
729  const FEDHeader fedHeader(event + eventSize);
730  const uint16_t fedId = fedHeader.sourceID();
732  throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection") << "Out of range FED ID : " << fedId;
733  } else if (fedId >= MINTCDSuTCAFEDID_ && fedId <= MAXTCDSuTCAFEDID_) {
734  if (!selectedTCDSFed) {
735  selectedTCDSFed = fedId;
736  tcds_pointer_ = event + eventSize;
738  tcdsInRange = true;
739  }
740  } else
741  throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection")
742  << "Second TCDS FED ID " << fedId << " found. First ID: " << selectedTCDSFed;
743  }
745  if (evf::evtn::evm_board_sense(event + eventSize, fedSize))
746  GTPEventID_ = evf::evtn::get(event + eventSize, true);
747  else
748  GTPEventID_ = evf::evtn::get(event + eventSize, false);
749  //evf::evtn::evm_board_setformat(fedSize);
750  const uint64_t gpsl = evf::evtn::getgpslow(event + eventSize);
751  const uint64_t gpsh = evf::evtn::getgpshigh(event + eventSize);
752  tstamp = edm::Timestamp(static_cast<edm::TimeValue_t>((gpsh << 32) + gpsl));
753  }
754  //take event ID from GTPE FED
756  if (evf::evtn::gtpe_board_sense(event + eventSize)) {
757  GTPEventID_ = evf::evtn::gtpe_get(event + eventSize);
758  }
759  }
760  FEDRawData& fedData = rawData.FEDData(fedId);
761  fedData.resize(fedSize);
762  memcpy(fedData.data(), event + eventSize, fedSize);
763  }
764  assert(eventSize == 0);
765 
766  return tstamp;
767 }
768 
770 
772  bool stop = false;
773  unsigned int currentLumiSection = 0;
774 
775  {
776  std::unique_lock<std::mutex> lk(startupLock_);
777  startupCv_.notify_one();
778  }
779 
780  uint32_t ls = 0;
781  uint32_t monLS = 1;
782  uint32_t lockCount = 0;
783  uint64_t sumLockWaitTimeUs = 0.;
784 
785  while (!stop) {
786  //wait for at least one free thread and chunk
787  int counter = 0;
788 
789  while ((workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty() ||
791  //report state to monitoring
792  if (fms_) {
793  bool copy_active = false;
794  for (auto j : tid_active_)
795  if (j)
796  copy_active = true;
799  else if (freeChunks_.empty()) {
800  if (copy_active)
802  else
804  } else {
805  if (copy_active)
807  else
809  }
810  }
811  std::unique_lock<std::mutex> lkw(mWakeup_);
812  //sleep until woken up by condition or a timeout
813  if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout) {
814  counter++;
815  if (!(counter % 6000)) {
816  edm::LogWarning("FedRawDataInputSource")
817  << "No free chunks or threads. Worker pool empty:" << workerPool_.empty()
818  << ", free chunks empty:" << freeChunks_.empty() << ", number of files buffered:" << readingFilesCount_
819  << " / " << maxBufferedFiles_;
820  }
821  LogDebug("FedRawDataInputSource") << "No free chunks or threads...";
822  } else {
823  assert(!(workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty());
824  }
825  if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
826  stop = true;
827  break;
828  }
829  }
830  //if this is reached, there are enough buffers and threads to proceed or processing is instructed to stop
831 
832  if (stop)
833  break;
834 
835  //look for a new file
836  std::string nextFile;
837  uint32_t fileSizeIndex;
838  int64_t fileSizeFromMetadata;
839 
840  if (fms_) {
843  }
844 
846  uint16_t rawHeaderSize = 0;
847  uint32_t lsFromRaw = 0;
848  int32_t serverEventsInNewFile = -1;
849  int rawFd = -1;
850 
851  int backoff_exp = 0;
852 
853  //entering loop which tries to grab new file from ramdisk
855  //check if hltd has signalled to throttle input
856  counter = 0;
857  while (daqDirector_->inputThrottled()) {
858  if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed))
859  break;
860 
861  unsigned int nConcurrentLumis = daqDirector_->numConcurrentLumis();
862  unsigned int nOtherLumis = nConcurrentLumis > 0 ? nConcurrentLumis - 1 : 0;
863  unsigned int checkLumiStart = currentLumiSection > nOtherLumis ? currentLumiSection - nOtherLumis : 1;
864  bool hasDiscardedLumi = false;
865  for (unsigned int i = checkLumiStart; i <= currentLumiSection; i++) {
867  edm::LogWarning("FedRawDataInputSource") << "Source detected that the lumisection is discarded -: " << i;
868  hasDiscardedLumi = true;
869  break;
870  }
871  }
872  if (hasDiscardedLumi)
873  break;
874 
876 
877  if (!(counter % 50))
878  edm::LogWarning("FedRawDataInputSource") << "Input throttled detected, reading files is paused...";
879  usleep(100000);
880  counter++;
881  }
882 
883  if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
884  stop = true;
885  break;
886  }
887 
888  assert(rawFd == -1);
889  uint64_t thisLockWaitTimeUs = 0.;
891  if (fileListMode_) {
892  //return LS if LS not set, otherwise return file
893  status = getFile(ls, nextFile, fileSizeIndex, thisLockWaitTimeUs);
896  rawFd,
897  rawHeaderSize,
898  lsFromRaw,
899  serverEventsInNewFile,
900  fileSizeFromMetadata,
901  false,
902  false,
903  false) != 0) {
904  //error
905  setExceptionState_ = true;
906  stop = true;
907  break;
908  }
909  if (!getLSFromFilename_)
910  ls = lsFromRaw;
911  }
912  } else if (!useFileBroker_)
914  ls, nextFile, fileSizeIndex, rawHeaderSize, thisLockWaitTimeUs, setExceptionState_);
915  else {
916  status = daqDirector_->getNextFromFileBroker(currentLumiSection,
917  ls,
918  nextFile,
919  rawFd,
920  rawHeaderSize,
921  serverEventsInNewFile,
922  fileSizeFromMetadata,
923  thisLockWaitTimeUs);
924  }
925 
927 
928  //cycle through all remaining LS even if no files get assigned
929  if (currentLumiSection != ls && status == evf::EvFDaqDirector::runEnded)
931 
932  //monitoring of lock wait time
933  if (thisLockWaitTimeUs > 0.)
934  sumLockWaitTimeUs += thisLockWaitTimeUs;
935  lockCount++;
936  if (ls > monLS) {
937  monLS = ls;
938  if (lockCount)
939  if (fms_)
940  fms_->reportLockWait(monLS, sumLockWaitTimeUs, lockCount);
941  lockCount = 0;
942  sumLockWaitTimeUs = 0;
943  }
944 
945  //check again for any remaining index/EoLS files after EoR file is seen
948  usleep(100000);
949  //now all files should have appeared in ramdisk, check again if any raw files were left behind
951  ls, nextFile, fileSizeIndex, rawHeaderSize, thisLockWaitTimeUs, setExceptionState_);
952  if (currentLumiSection != ls && status == evf::EvFDaqDirector::runEnded)
954  }
955 
957  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runEnded));
958  fileQueue_.push(std::move(inf));
959  stop = true;
960  break;
961  }
962 
963  //error from filelocking function
965  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runAbort, 0));
966  fileQueue_.push(std::move(inf));
967  stop = true;
968  break;
969  }
970  //queue new lumisection
971  if (getLSFromFilename_) {
972  if (ls > currentLumiSection) {
973  if (!useFileBroker_) {
974  //file locking
975  //setMonStateSup(inSupNewLumi);
976  currentLumiSection = ls;
977  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, currentLumiSection));
978  fileQueue_.push(std::move(inf));
979  } else {
980  //new file service
981  if (currentLumiSection == 0 && !alwaysStartFromFirstLS_) {
983  //start transitions from LS specified by env, continue if not reached
984  if (ls < daqDirector_->getStartLumisectionFromEnv()) {
985  //skip file if from earlier LS than specified by env
986  if (rawFd != -1) {
987  close(rawFd);
988  rawFd = -1;
989  }
991  continue;
992  } else {
993  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, ls));
994  fileQueue_.push(std::move(inf));
995  }
996  } else if (ls < 100) {
997  //look at last LS file on disk to start from that lumisection (only within first 100 LS)
998  unsigned int lsToStart = daqDirector_->getLumisectionToStart();
999 
1000  for (unsigned int nextLS = std::min(lsToStart, ls); nextLS <= ls; nextLS++) {
1001  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, nextLS));
1002  fileQueue_.push(std::move(inf));
1003  }
1004  } else {
1005  //start from current LS
1006  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, ls));
1007  fileQueue_.push(std::move(inf));
1008  }
1009  } else {
1010  //queue all lumisections after last one seen to avoid gaps
1011  for (unsigned int nextLS = currentLumiSection + 1; nextLS <= ls; nextLS++) {
1012  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, nextLS));
1013  fileQueue_.push(std::move(inf));
1014  }
1015  }
1016  currentLumiSection = ls;
1017  }
1018  }
1019  //else
1020  if (currentLumiSection > 0 && ls < currentLumiSection) {
1021  edm::LogError("FedRawDataInputSource")
1022  << "Got old LS (" << ls << ") file from EvFDAQDirector! Expected LS:" << currentLumiSection
1023  << ". Aborting execution." << std::endl;
1024  if (rawFd != -1)
1025  close(rawFd);
1026  rawFd = -1;
1027  std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runAbort, 0));
1028  fileQueue_.push(std::move(inf));
1029  stop = true;
1030  break;
1031  }
1032  }
1033 
1034  int dbgcount = 0;
1037  dbgcount++;
1038  if (!(dbgcount % 20))
1039  LogDebug("FedRawDataInputSource") << "No file for me... sleep and try again...";
1040  if (!useFileBroker_)
1041  usleep(100000);
1042  else {
1043  backoff_exp = std::min(4, backoff_exp); // max 1.6 seconds
1044  //backoff_exp=0; // disabled!
1045  int sleeptime = (int)(100000. * pow(2, backoff_exp));
1046  usleep(sleeptime);
1047  backoff_exp++;
1048  }
1049  } else
1050  backoff_exp = 0;
1051  }
1052  //end of file grab loop, parse result
1055  LogDebug("FedRawDataInputSource") << "The director says to grab -: " << nextFile;
1056 
1057  std::string rawFile;
1058  //file service will report raw extension
1059  if (useFileBroker_ || rawHeaderSize)
1060  rawFile = nextFile;
1061  else {
1062  std::filesystem::path rawFilePath(nextFile);
1063  rawFile = rawFilePath.replace_extension(".raw").string();
1064  }
1065 
1066  struct stat st;
1067  int stat_res = stat(rawFile.c_str(), &st);
1068  if (stat_res == -1) {
1069  edm::LogError("FedRawDataInputSource") << "Can not stat file (" << errno << "):-" << rawFile << std::endl;
1070  setExceptionState_ = true;
1071  break;
1072  }
1073  uint64_t fileSize = st.st_size;
1074 
1075  if (fms_) {
1079  }
1080  int eventsInNewFile;
1081  if (fileListMode_) {
1082  if (fileSize == 0)
1083  eventsInNewFile = 0;
1084  else
1085  eventsInNewFile = -1;
1086  } else {
1088  if (!useFileBroker_) {
1089  if (rawHeaderSize) {
1090  int rawFdEmpty = -1;
1091  uint16_t rawHeaderCheck;
1092  bool fileFound;
1093  eventsInNewFile = daqDirector_->grabNextJsonFromRaw(
1094  nextFile, rawFdEmpty, rawHeaderCheck, fileSizeFromMetadata, fileFound, 0, true);
1095  assert(fileFound && rawHeaderCheck == rawHeaderSize);
1097  } else
1098  eventsInNewFile = daqDirector_->grabNextJsonFileAndUnlock(nextFile);
1099  } else
1100  eventsInNewFile = serverEventsInNewFile;
1101  assert(eventsInNewFile >= 0);
1102  assert((eventsInNewFile > 0) ==
1103  (fileSize > rawHeaderSize)); //file without events must be empty or contain only header
1104  }
1105 
1106  if (!singleBufferMode_) {
1107  //calculate number of needed chunks
1108  unsigned int neededChunks = fileSize / eventChunkSize_;
1109  if (fileSize % eventChunkSize_)
1110  neededChunks++;
1111 
1112  std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1113  ls,
1114  rawFile,
1115  !fileListMode_,
1116  rawFd,
1117  fileSize,
1118  rawHeaderSize,
1119  neededChunks,
1120  eventsInNewFile,
1121  this));
1123  auto newInputFilePtr = newInputFile.get();
1124  fileQueue_.push(std::move(newInputFile));
1125 
1126  for (unsigned int i = 0; i < neededChunks; i++) {
1127  if (fms_) {
1128  bool copy_active = false;
1129  for (auto j : tid_active_)
1130  if (j)
1131  copy_active = true;
1132  if (copy_active)
1134  else
1136  }
1137  //get thread
1138  unsigned int newTid = 0xffffffff;
1139  while (!workerPool_.try_pop(newTid)) {
1140  usleep(100000);
1141  if (quit_threads_.load(std::memory_order_relaxed)) {
1142  stop = true;
1143  break;
1144  }
1145  }
1146 
1147  if (fms_) {
1148  bool copy_active = false;
1149  for (auto j : tid_active_)
1150  if (j)
1151  copy_active = true;
1152  if (copy_active)
1154  else
1156  }
1157  InputChunk* newChunk = nullptr;
1158  while (!freeChunks_.try_pop(newChunk)) {
1159  usleep(100000);
1160  if (quit_threads_.load(std::memory_order_relaxed)) {
1161  stop = true;
1162  break;
1163  }
1164  }
1165 
1166  if (newChunk == nullptr) {
1167  //return unused tid if we received shutdown (nullptr chunk)
1168  if (newTid != 0xffffffff)
1169  workerPool_.push(newTid);
1170  stop = true;
1171  break;
1172  }
1173  if (stop)
1174  break;
1176 
1177  std::unique_lock<std::mutex> lk(mReader_);
1178 
1179  unsigned int toRead = eventChunkSize_;
1180  if (i == neededChunks - 1 && fileSize % eventChunkSize_)
1181  toRead = fileSize % eventChunkSize_;
1182  newChunk->reset(i * eventChunkSize_, toRead, i);
1183 
1184  workerJob_[newTid].first = newInputFilePtr;
1185  workerJob_[newTid].second = newChunk;
1186 
1187  //wake up the worker thread
1188  cvReader_[newTid]->notify_one();
1189  }
1190  } else {
1191  if (!eventsInNewFile) {
1192  if (rawFd) {
1193  close(rawFd);
1194  rawFd = -1;
1195  }
1196  //still queue file for lumi update
1197  std::unique_lock<std::mutex> lkw(mWakeup_);
1198  //TODO: also file with only file header fits in this edge case. Check if read correctly in single buffer mode
1199  std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1200  ls,
1201  rawFile,
1202  !fileListMode_,
1203  rawFd,
1204  fileSize,
1205  rawHeaderSize,
1206  (rawHeaderSize > 0),
1207  0,
1208  this));
1210  fileQueue_.push(std::move(newInputFile));
1211  cvWakeup_.notify_one();
1212  break;
1213  }
1214  //in single-buffer mode put single chunk in the file and let the main thread read the file
1215  InputChunk* newChunk;
1216  //should be available immediately
1217  while (!freeChunks_.try_pop(newChunk)) {
1218  usleep(100000);
1219  if (quit_threads_.load(std::memory_order_relaxed))
1220  break;
1221  }
1222 
1223  std::unique_lock<std::mutex> lkw(mWakeup_);
1224 
1225  unsigned int toRead = eventChunkSize_;
1226  if (fileSize % eventChunkSize_)
1227  toRead = fileSize % eventChunkSize_;
1228  newChunk->reset(0, toRead, 0);
1229  newChunk->readComplete_ = true;
1230 
1231  //push file and wakeup main thread
1232  std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
1233  ls,
1234  rawFile,
1235  !fileListMode_,
1236  rawFd,
1237  fileSize,
1238  rawHeaderSize,
1239  1,
1240  eventsInNewFile,
1241  this));
1242  newInputFile->chunks_[0] = newChunk;
1244  fileQueue_.push(std::move(newInputFile));
1245  cvWakeup_.notify_one();
1246  }
1247  }
1248  }
1250  //make sure threads finish reading
1251  unsigned numFinishedThreads = 0;
1252  while (numFinishedThreads < workerThreads_.size()) {
1253  unsigned tid = 0;
1254  while (!workerPool_.try_pop(tid)) {
1255  usleep(10000);
1256  }
1257  std::unique_lock<std::mutex> lk(mReader_);
1258  thread_quit_signal[tid] = true;
1259  cvReader_[tid]->notify_one();
1260  numFinishedThreads++;
1261  }
1262  for (unsigned int i = 0; i < workerThreads_.size(); i++) {
1263  workerThreads_[i]->join();
1264  delete workerThreads_[i];
1265  }
1266 }
1267 
1268 void FedRawDataInputSource::readWorker(unsigned int tid) {
1269  bool init = true;
1270 
1271  while (true) {
1272  tid_active_[tid] = false;
1273  std::unique_lock<std::mutex> lk(mReader_);
1274  workerJob_[tid].first = nullptr;
1275  workerJob_[tid].first = nullptr;
1276 
1277  assert(!thread_quit_signal[tid]); //should never get it here
1278  workerPool_.push(tid);
1279 
1280  if (init) {
1281  std::unique_lock<std::mutex> lk(startupLock_);
1282  init = false;
1283  startupCv_.notify_one();
1284  }
1285  cvReader_[tid]->wait(lk);
1286 
1287  if (thread_quit_signal[tid])
1288  return;
1289  tid_active_[tid] = true;
1290 
1291  InputFile* file;
1292  InputChunk* chunk;
1293 
1294  assert(workerJob_[tid].first != nullptr && workerJob_[tid].second != nullptr);
1295 
1296  file = workerJob_[tid].first;
1297  chunk = workerJob_[tid].second;
1298 
1299  //skip reading initial header size in first chunk if inheriting file descriptor (already set at appropriate position)
1300  unsigned int bufferLeft = (chunk->offset_ == 0 && file->rawFd_ != -1) ? file->rawHeaderSize_ : 0;
1301 
1302  //if only one worker thread exists, use single fd for all operations
1303  //if more worker threads exist, use rawFd_ for only the first read operation and then close file
1304  int fileDescriptor;
1305  bool fileOpenedHere = false;
1306 
1307  if (numConcurrentReads_ == 1) {
1308  fileDescriptor = file->rawFd_;
1309  if (fileDescriptor == -1) {
1310  fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
1311  fileOpenedHere = true;
1312  file->rawFd_ = fileDescriptor;
1313  }
1314  } else {
1315  if (chunk->offset_ == 0) {
1316  fileDescriptor = file->rawFd_;
1317  file->rawFd_ = -1;
1318  if (fileDescriptor == -1) {
1319  fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
1320  fileOpenedHere = true;
1321  }
1322  } else {
1323  fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
1324  fileOpenedHere = true;
1325  }
1326  }
1327 
1328  if (fileDescriptor < 0) {
1329  edm::LogError("FedRawDataInputSource") << "readWorker failed to open file -: " << file->fileName_
1330  << " fd:" << fileDescriptor << " error: " << strerror(errno);
1331  setExceptionState_ = true;
1332  continue;
1333  }
1334  if (fileOpenedHere) { //fast forward to this chunk position
1335  off_t pos = 0;
1336  pos = lseek(fileDescriptor, chunk->offset_, SEEK_SET);
1337  if (pos == -1) {
1338  edm::LogError("FedRawDataInputSource")
1339  << "readWorker failed to seek file -: " << file->fileName_ << " fd:" << fileDescriptor << " to offset "
1340  << chunk->offset_ << " error: " << strerror(errno);
1341  setExceptionState_ = true;
1342  continue;
1343  }
1344  }
1345 
1346  LogDebug("FedRawDataInputSource") << "Reader thread opened file -: TID: " << tid << " file: " << file->fileName_
1347  << " at offset " << lseek(fileDescriptor, 0, SEEK_CUR);
1348 
1349  unsigned int skipped = bufferLeft;
1351  for (unsigned int i = 0; i < readBlocks_; i++) {
1352  ssize_t last;
1353 
1354  //protect against reading into next block
1355  last = ::read(
1356  fileDescriptor, (void*)(chunk->buf_ + bufferLeft), std::min(chunk->usedSize_ - bufferLeft, eventChunkBlock_));
1357 
1358  if (last < 0) {
1359  edm::LogError("FedRawDataInputSource") << "readWorker failed to read file -: " << file->fileName_
1360  << " fd:" << fileDescriptor << " error: " << strerror(errno);
1361  setExceptionState_ = true;
1362  break;
1363  }
1364  if (last > 0)
1365  bufferLeft += last;
1366  if (last < eventChunkBlock_) { //last read
1367  //check if this is last block, then total read size must match file size
1368  if (!(chunk->usedSize_ - skipped == i * eventChunkBlock_ + last)) {
1369  edm::LogError("FedRawDataInputSource")
1370  << "readWorker failed to read file -: " << file->fileName_ << " fd:" << fileDescriptor << " last:" << last
1371  << " expectedChunkSize:" << chunk->usedSize_
1372  << " readChunkSize:" << (skipped + i * eventChunkBlock_ + last) << " skipped:" << skipped
1373  << " block:" << (i + 1) << "/" << readBlocks_ << " error: " << strerror(errno);
1374  setExceptionState_ = true;
1375  }
1376  break;
1377  }
1378  }
1379  if (setExceptionState_)
1380  continue;
1381 
1383  auto diff = end - start;
1384  std::chrono::milliseconds msec = std::chrono::duration_cast<std::chrono::milliseconds>(diff);
1385  LogDebug("FedRawDataInputSource") << " finished reading block -: " << (bufferLeft >> 20) << " MB"
1386  << " in " << msec.count() << " ms (" << (bufferLeft >> 20) / double(msec.count())
1387  << " GB/s)";
1388 
1389  if (chunk->offset_ + bufferLeft == file->fileSize_) { //file reading finished using same fd
1390  close(fileDescriptor);
1391  fileDescriptor = -1;
1392  if (numConcurrentReads_ == 1)
1393  file->rawFd_ = -1;
1394  }
1395  if (numConcurrentReads_ > 1 && fileDescriptor != -1)
1396  close(fileDescriptor);
1397 
1398  //detect FRD event version. Skip file Header if it exists
1399  if (detectedFRDversion_ == 0 && chunk->offset_ == 0) {
1400  detectedFRDversion_ = *((uint16_t*)(chunk->buf_ + file->rawHeaderSize_));
1401  }
1403  chunk->readComplete_ =
1404  true; //this is atomic to secure the sequential buffer fill before becoming available for processing)
1405  file->chunks_[chunk->fileIndex_] = chunk; //put the completed chunk in the file chunk vector at predetermined index
1406  }
1407 }
1408 
1410  quit_threads_ = true;
1411  throw cms::Exception("FedRawDataInputSource:threadError") << " file reader thread error ";
1412 }
1413 
1415  if (fms_)
1416  fms_->setInState(state);
1417 }
1418 
1420  if (fms_)
1422 }
1423 
1424 inline bool InputFile::advance(unsigned char*& dataPosition, const size_t size) {
1425  //wait for chunk
1426 
1427  while (!waitForChunk(currentChunk_)) {
1429  usleep(100000);
1431  if (parent_->exceptionState())
1432  parent_->threadError();
1433  }
1434 
1435  dataPosition = chunks_[currentChunk_]->buf_ + chunkPosition_;
1436  size_t currentLeft = chunks_[currentChunk_]->size_ - chunkPosition_;
1437 
1438  if (currentLeft < size) {
1439  //we need next chunk
1440  assert(chunks_.size() > currentChunk_ + 1);
1441  while (!waitForChunk(currentChunk_ + 1)) {
1443  usleep(100000);
1445  if (parent_->exceptionState())
1446  parent_->threadError();
1447  }
1448  //copy everything to beginning of the first chunk
1449  dataPosition -= chunkPosition_;
1450  assert(dataPosition == chunks_[currentChunk_]->buf_);
1451  memmove(chunks_[currentChunk_]->buf_, chunks_[currentChunk_]->buf_ + chunkPosition_, currentLeft);
1452  memcpy(chunks_[currentChunk_]->buf_ + currentLeft, chunks_[currentChunk_ + 1]->buf_, size - currentLeft);
1453  //set pointers at the end of the old data position
1454  bufferPosition_ += size;
1455  chunkPosition_ = size - currentLeft;
1456  currentChunk_++;
1457  return true;
1458  } else {
1459  chunkPosition_ += size;
1460  bufferPosition_ += size;
1461  return false;
1462  }
1463 }
1464 
1465 inline void InputFile::moveToPreviousChunk(const size_t size, const size_t offset) {
1466  //this will fail in case of events that are too large
1468  assert(size - offset < chunks_[currentChunk_]->size_);
1469  memcpy(chunks_[currentChunk_ - 1]->buf_ + offset, chunks_[currentChunk_]->buf_ + chunkPosition_, size);
1470  chunkPosition_ += size;
1471  bufferPosition_ += size;
1472 }
1473 
1474 inline void InputFile::rewindChunk(const size_t size) {
1475  chunkPosition_ -= size;
1476  bufferPosition_ -= size;
1477 }
1478 
1480  if (rawFd_ != -1)
1481  close(rawFd_);
1482 
1483  if (deleteFile_ && !fileName_.empty()) {
1485  try {
1486  //sometimes this fails but file gets deleted
1487  LogDebug("FedRawDataInputSource:InputFile") << "Deleting input file -:" << fileName_;
1489  return;
1490  } catch (const std::filesystem::filesystem_error& ex) {
1491  edm::LogError("FedRawDataInputSource:InputFile")
1492  << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what() << ". Trying again.";
1493  } catch (std::exception& ex) {
1494  edm::LogError("FedRawDataInputSource:InputFile")
1495  << " - deleteFile std::exception CAUGHT -: " << ex.what() << ". Trying again.";
1496  }
1498  }
1499 }
1500 
1501 //single-buffer mode file reading
1503  uint32_t existingSize = 0;
1504 
1505  if (fileDescriptor_ < 0) {
1506  bufferInputRead_ = 0;
1507  if (file->rawFd_ == -1) {
1508  fileDescriptor_ = open(file->fileName_.c_str(), O_RDONLY);
1509  if (file->rawHeaderSize_)
1510  lseek(fileDescriptor_, file->rawHeaderSize_, SEEK_SET);
1511  } else
1512  fileDescriptor_ = file->rawFd_;
1513 
1514  //skip header size in destination buffer (chunk position was already adjusted)
1515  bufferInputRead_ += file->rawHeaderSize_;
1516  existingSize += file->rawHeaderSize_;
1517 
1518  if (fileDescriptor_ >= 0)
1519  LogDebug("FedRawDataInputSource") << "opened file -: " << std::endl << file->fileName_;
1520  else {
1521  throw cms::Exception("FedRawDataInputSource:readNextChunkIntoBuffer")
1522  << "failed to open file " << std::endl
1523  << file->fileName_ << " fd:" << fileDescriptor_;
1524  }
1525  //fill chunk (skipping file header if present)
1526  for (unsigned int i = 0; i < readBlocks_; i++) {
1527  const ssize_t last = ::read(fileDescriptor_,
1528  (void*)(file->chunks_[0]->buf_ + existingSize),
1529  eventChunkBlock_ - (i == readBlocks_ - 1 ? existingSize : 0));
1531  existingSize += last;
1532  }
1533 
1534  } else {
1535  //continue reading
1536  if (file->chunkPosition_ == 0) { //in the rare case the last byte barely fit
1537  for (unsigned int i = 0; i < readBlocks_; i++) {
1538  const ssize_t last = ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSize), eventChunkBlock_);
1540  existingSize += last;
1541  }
1542  } else {
1543  //event didn't fit in last chunk, so leftover must be moved to the beginning and completed
1544  uint32_t existingSizeLeft = eventChunkSize_ - file->chunkPosition_;
1545  memmove((void*)file->chunks_[0]->buf_, file->chunks_[0]->buf_ + file->chunkPosition_, existingSizeLeft);
1546 
1547  //calculate amount of data that can be added
1548  const uint32_t blockcount = file->chunkPosition_ / eventChunkBlock_;
1549  const uint32_t leftsize = file->chunkPosition_ % eventChunkBlock_;
1550 
1551  for (uint32_t i = 0; i < blockcount; i++) {
1552  const ssize_t last =
1553  ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSizeLeft), eventChunkBlock_);
1555  existingSizeLeft += last;
1556  }
1557  if (leftsize) {
1558  const ssize_t last = ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSizeLeft), leftsize);
1560  }
1561  file->chunkPosition_ = 0; //data was moved to beginning of the chunk
1562  }
1563  }
1564  if (bufferInputRead_ == file->fileSize_) { // no more data in this file
1565  if (fileDescriptor_ != -1) {
1566  LogDebug("FedRawDataInputSource") << "Closing input file -: " << std::endl << file->fileName_;
1567  close(fileDescriptor_);
1568  file->rawFd_ = fileDescriptor_ = -1;
1569  }
1570  }
1571 }
1572 
1574  std::lock_guard<std::mutex> lock(monlock_);
1575  auto itr = sourceEventsReport_.find(lumi);
1576  if (itr != sourceEventsReport_.end())
1577  itr->second += events;
1578  else
1580 }
1581 
1582 std::pair<bool, unsigned int> FedRawDataInputSource::getEventReport(unsigned int lumi, bool erase) {
1583  std::lock_guard<std::mutex> lock(monlock_);
1584  auto itr = sourceEventsReport_.find(lumi);
1585  if (itr != sourceEventsReport_.end()) {
1586  std::pair<bool, unsigned int> ret(true, itr->second);
1587  if (erase)
1588  sourceEventsReport_.erase(itr);
1589  return ret;
1590  } else
1591  return std::pair<bool, unsigned int>(false, 0);
1592 }
1593 
1595  std::sort(fileNames_.begin(), fileNames_.end(), [](std::string a, std::string b) {
1596  if (a.rfind('/') != std::string::npos)
1597  a = a.substr(a.rfind('/'));
1598  if (b.rfind('/') != std::string::npos)
1599  b = b.substr(b.rfind('/'));
1600  return b > a;
1601  });
1602 
1603  if (!fileNames_.empty()) {
1604  //get run number from first file in the vector
1606  std::string fileStem = fileName.stem().string();
1607  if (fileStem.find("file://") == 0)
1608  fileStem = fileStem.substr(7);
1609  else if (fileStem.find("file:") == 0)
1610  fileStem = fileStem.substr(5);
1611  auto end = fileStem.find('_');
1612 
1613  if (fileStem.find("run") == 0) {
1614  std::string runStr = fileStem.substr(3, end - 3);
1615  try {
1616  //get long to support test run numbers < 2^32
1617  long rval = std::stol(runStr);
1618  edm::LogInfo("FedRawDataInputSource") << "Autodetected run number in fileListMode -: " << rval;
1619  return rval;
1620  } catch (const std::exception&) {
1621  edm::LogWarning("FedRawDataInputSource")
1622  << "Unable to autodetect run number in fileListMode from file -: " << fileName;
1623  }
1624  }
1625  }
1626  return -1;
1627 }
1628 
1630  std::string& nextFile,
1631  uint32_t& fsize,
1632  uint64_t& lockWaitTime) {
1633  if (fileListIndex_ < fileNames_.size()) {
1634  nextFile = fileNames_[fileListIndex_];
1635  if (nextFile.find("file://") == 0)
1636  nextFile = nextFile.substr(7);
1637  else if (nextFile.find("file:") == 0)
1638  nextFile = nextFile.substr(5);
1639  std::filesystem::path fileName = nextFile;
1640  std::string fileStem = fileName.stem().string();
1641  if (fileStem.find("ls"))
1642  fileStem = fileStem.substr(fileStem.find("ls") + 2);
1643  if (fileStem.find('_'))
1644  fileStem = fileStem.substr(0, fileStem.find('_'));
1645 
1646  if (!fileListLoopMode_)
1647  ls = std::stoul(fileStem);
1648  else //always starting from LS 1 in loop mode
1649  ls = 1 + loopModeIterationInc_;
1650 
1651  //fsize = 0;
1652  //lockWaitTime = 0;
1653  fileListIndex_++;
1655  } else {
1656  if (!fileListLoopMode_)
1658  else {
1659  //loop through files until interrupted
1661  fileListIndex_ = 0;
1662  return getFile(ls, nextFile, fsize, lockWaitTime);
1663  }
1664  }
1665 }
size
Write out results.
static const char runNumber_[]
uint8_t triggerType() const
Event Trigger type identifier.
Definition: FEDHeader.cc:13
Definition: start.py:1
Definition: fillJson.h:27
std::vector< std::string > fileNames_
unsigned int getgpshigh(const unsigned char *)
uint32_t chunkPosition_
std::condition_variable cvWakeup_
void read(edm::EventPrincipal &eventPrincipal) override
std::shared_ptr< RunAuxiliary > runAuxiliary() const
Called by the framework to merge or insert run in principal cache.
Definition: InputSource.h:232
ProductRegistry & productRegistryUpdate()
Definition: InputSource.h:330
bool gtpe_board_sense(const unsigned char *p)
constexpr size_t FRDHeaderMaxVersion
void setExceptionDetected(unsigned int ls)
static Timestamp invalidTimestamp()
Definition: Timestamp.h:75
bool crc32c_hw_test()
Definition: crc32c.cc:354
unsigned int offset_
std::pair< bool, unsigned int > getEventReport(unsigned int lumi, bool erase)
static void fillDescriptions(edm::ConfigurationDescriptions &descriptions)
static const uint32_t length
Definition: FEDTrailer.h:57
tbb::concurrent_queue< unsigned int > workerPool_
unsigned int get(const unsigned char *, bool)
bool useFileBroker() const
void maybeOpenNewLumiSection(const uint32_t lumiSection)
static const uint32_t length
Definition: FEDHeader.h:54
ret
prodAgent to be discontinued
uint16_t sourceID() const
Identifier of the FED.
Definition: FEDHeader.cc:19
evf::EvFDaqDirector::FileStatus getFile(unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint64_t &lockWaitTime)
void rewindChunk(const size_t size)
volatile std::atomic< bool > shutdown_flag
RunNumber_t run() const
Accessor for current run number.
Definition: InputSource.cc:457
tbb::concurrent_queue< std::unique_ptr< InputFile > > fileQueue_
std::string getEoLSFilePathOnFU(const unsigned int ls) const
std::atomic< bool > quit_threads_
std::vector< ReaderInfo > workerJob_
bool lumisectionDiscarded(unsigned int ls)
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)
edm::Timestamp fillFEDRawDataCollection(FEDRawDataCollection &rawData, bool &tcdsInRange)
std::pair< InputFile *, InputChunk * > ReaderInfo
std::unique_ptr< std::thread > readSupervisorThread_
Log< level::Error, false > LogError
unsigned int numConcurrentLumis() const
StreamID streamID() const
int timeout
Definition: mps_check.py:53
assert(be >=bs)
bool isExceptionOnData(unsigned int ls)
ProcessHistoryID daqInit(ProductRegistry &productRegistry, ProcessHistoryRegistry &processHistoryRegistry) const
FedRawDataInputSource * parent_
std::string getEoRFilePathOnFU() const
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)
uint32_t fragmentLength() const
The length of the event fragment counted in 64-bit words including header and trailer.
Definition: FEDTrailer.cc:13
FedRawDataInputSource(edm::ParameterSet const &, edm::InputSourceDescription const &)
U second(std::pair< T, U > const &p)
std::list< std::pair< int, std::unique_ptr< InputFile > > > filesToDelete_
void reset(unsigned int newOffset, unsigned int toRead, unsigned int fileIndex)
std::string getEoLSFilePathOnBU(const unsigned int ls) const
static Timestamp beginOfTime()
Definition: Timestamp.h:77
void put(BranchDescription const &bd, std::unique_ptr< WrapperBase > edp, ProductProvenance const &productProvenance) const
const std::vector< unsigned int > testTCDSFEDRange_
uint32_t bufferPosition_
Definition: TCDSRaw.h:16
std::string const & processGUID() const
Accessor for global process identifier.
Definition: InputSource.h:197
void resize(size_t newsize)
Definition: FEDRawData.cc:28
bool advance(unsigned char *&dataPosition, const size_t size)
void setLuminosityBlockAuxiliary(LuminosityBlockAuxiliary *lbp)
Definition: InputSource.h:337
std::vector< int > streamFileTracker_
void moveToPreviousChunk(const size_t size, const size_t offset)
unsigned char * buf_
ProductProvenance const & dummyProvenance() const
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)
void setMonStateSup(evf::FastMonState::InputState state)
int grabNextJsonFileAndUnlock(std::filesystem::path const &jsonSourcePath)
std::map< unsigned int, unsigned int > sourceEventsReport_
edm::ProcessHistoryID processHistoryID_
bool waitForChunk(unsigned int chunkid)
const edm::DaqProvenanceHelper daqProvenanceHelper_
unsigned int fileIndex_
unsigned long long TimeValue_t
Definition: Timestamp.h:21
std::vector< std::thread * > workerThreads_
bool evm_board_sense(const unsigned char *p, size_t size)
void setMonState(evf::FastMonState::InputState state)
void setEventCached()
Called by the framework to merge or ached() const {return eventCached_;}.
Definition: InputSource.h:362
void createProcessingNotificationMaybe() const
uint32_t crc32c(uint32_t crc, const unsigned char *buf, size_t len)
Definition: crc32c.cc:340
void setInputSource(FedRawDataInputSource *inputSource)
Log< level::Info, false > LogInfo
std::atomic< bool > readComplete_
Definition: init.py:1
FileStatus updateFuLock(unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint16_t &rawHeaderSize, uint64_t &lockWaitTime, bool &setExceptionState)
def ls(path, rec=False)
Definition: eostools.py:349
BranchDescription const & branchDescription() const
unsigned int getStartLumisectionFromEnv() const
int grabNextJsonFromRaw(std::string const &rawSourcePath, int &rawFd, uint16_t &rawHeaderSize, int64_t &fileSizeFromHeader, bool &fileFound, uint32_t serverLS, bool closeFile)
def getRunNumber(filename)
void setInStateSup(FastMonState::InputState inputState)
unsigned long long uint64_t
Definition: Time.h:13
unsigned int currentChunk_
std::unique_ptr< FRDEventMsgView > event_
def load(fileName)
Definition: svgfig.py:547
std::string fileName_
double b
Definition: hdecay.h:118
void add(std::string const &label, ParameterSetDescription const &psetDescription)
def remove(d, key, TELL=False)
Definition: MatrixUtil.py:223
std::condition_variable startupCv_
void resetLuminosityBlockAuxiliary(bool isNewLumi=true) const
Definition: InputSource.h:345
void stoppedLookingForFile(unsigned int lumi)
tbb::concurrent_vector< InputChunk * > chunks_
ProcessHistoryRegistry & processHistoryRegistryForUpdate()
Definition: InputSource.h:331
ItemType state() const
Definition: InputSource.h:332
LuminosityBlockNumber_t luminosityBlock() const
Accessor for current luminosity block number.
Definition: InputSource.cc:462
HLT enums.
evf::EvFDaqDirector * daqDirector_
void setRunAuxiliary(RunAuxiliary *rp)
Definition: InputSource.h:333
void readWorker(unsigned int tid)
double a
Definition: hdecay.h:119
std::shared_ptr< LuminosityBlockAuxiliary > luminosityBlockAuxiliary() const
Called by the framework to merge or insert lumi in principal cache.
Definition: InputSource.h:235
void makeEvent(EventPrincipal &eventPrincipal, EventAuxiliary const &eventAuxiliary)
std::vector< unsigned int > thread_quit_signal
unsigned int getLumisectionToStart() const
std::atomic< unsigned int > readingFilesCount_
void reportEventsThisLumiInSource(unsigned int lumi, unsigned int events)
const unsigned char * data() const
Return a const pointer to the beginning of the data buffer.
Definition: FEDRawData.cc:24
std::vector< std::unique_ptr< std::condition_variable > > cvReader_
evf::FastMonitoringService * fms_
unsigned int gtpe_get(const unsigned char *)
std::vector< unsigned int > tid_active_
unsigned int RunNumber_t
void setDeleteTracking(std::mutex *fileDeleteLock, std::list< std::pair< int, std::unique_ptr< InputFile >>> *filesToDelete)
constexpr std::array< uint32, FRDHeaderMaxVersion+1 > FRDHeaderVersionSize
tbb::concurrent_queue< InputChunk * > freeChunks_
Log< level::Warning, false > LogWarning
evf::EvFDaqDirector::FileStatus getNextEvent()
evf::EvFDaqDirector::FileStatus nextEvent()
void readNextChunkIntoBuffer(InputFile *file)
void createBoLSFile(const uint32_t lumiSection, bool checkIfExists) const
void setFMS(evf::FastMonitoringService *fms)
std::unique_ptr< InputFile > currentFile_
unsigned int getgpslow(const unsigned char *)
int events
Power< A, B >::type pow(const A &a, const B &b)
Definition: Power.h:29
def move(src, dest)
Definition: eostools.py:511
void reportLockWait(unsigned int ls, double waitTime, unsigned int lockCount)
Definition: event.py:1
#define LogDebug(id)
void setInState(FastMonState::InputState inputState)