CMS 3D CMS Logo
List of all members | Public Member Functions | Static Public Member Functions | Protected Member Functions | Private Types | Private Member Functions | Private Attributes | Friends
FedRawDataInputSource Class Reference

#include <FedRawDataInputSource.h>

Inheritance diagram for FedRawDataInputSource:
edm::RawInputSource edm::InputSource

Public Member Functions

 FedRawDataInputSource (edm::ParameterSet const &, edm::InputSourceDescription const &)
 
std::pair< bool, unsigned int > getEventReport (unsigned int lumi, bool erase)
 
 ~FedRawDataInputSource () override
 
- Public Member Functions inherited from edm::RawInputSource
 RawInputSource (ParameterSet const &pset, InputSourceDescription const &desc)
 
 ~RawInputSource () override
 
- Public Member Functions inherited from edm::InputSource
std::shared_ptr< ActivityRegistry > actReg () const
 Accessor for Activity Registry. More...
 
std::shared_ptr< BranchIDListHelper > & branchIDListHelper ()
 
std::shared_ptr< BranchIDListHelper const > branchIDListHelper () const
 Accessors for branchIDListHelper. More...
 
void closeFile (FileBlock *, bool cleaningUpAfterException)
 close current file More...
 
void doBeginJob ()
 Called by framework at beginning of job. More...
 
virtual void doBeginLumi (LuminosityBlockPrincipal &lbp, ProcessContext const *)
 Called by framework at beginning of lumi block. More...
 
virtual void doBeginRun (RunPrincipal &rp, ProcessContext const *)
 Called by framework at beginning of run. More...
 
void doEndJob ()
 Called by framework at end of job. More...
 
void fillProcessBlockHelper ()
 Fill the ProcessBlockHelper with info for the current file. More...
 
ProcessingController::ForwardState forwardState () const
 
bool goToEvent (EventID const &eventID)
 
 InputSource (InputSource const &)=delete
 
 InputSource (ParameterSet const &, InputSourceDescription const &)
 Constructor. More...
 
void issueReports (EventID const &eventID, StreamID streamID)
 issue an event report More...
 
LuminosityBlockNumber_t luminosityBlock () const
 Accessor for current luminosity block number. More...
 
std::shared_ptr< LuminosityBlockAuxiliary > luminosityBlockAuxiliary () const
 Called by the framework to merge or insert lumi in principal cache. More...
 
int maxEvents () const
 
int maxLuminosityBlocks () const
 
ModuleDescription const & moduleDescription () const
 Accessor for 'module' description. More...
 
ItemType nextItemType ()
 Advances the source to the next item. More...
 
bool nextProcessBlock (ProcessBlockPrincipal &)
 Next process block, return false if there is none, sets the processName in the principal. More...
 
InputSource & operator= (InputSource const &)=delete
 
std::shared_ptr< ProcessBlockHelper > & processBlockHelper ()
 
std::shared_ptr< ProcessBlockHelper const > processBlockHelper () const
 Accessors for processBlockHelper. More...
 
ProcessConfiguration const & processConfiguration () const
 Accessor for Process Configuration. More...
 
std::string const & processGUID () const
 Accessor for global process identifier. More...
 
ProcessHistoryRegistry & processHistoryRegistry ()
 
ProcessHistoryRegistry const & processHistoryRegistry () const
 Accessors for process history registry. More...
 
ProcessingMode processingMode () const
 RunsLumisAndEvents (default), RunsAndLumis, or Runs. More...
 
std::shared_ptr< ProductRegistry > & productRegistry ()
 
std::shared_ptr< ProductRegistry const > productRegistry () const
 Accessors for product registry. More...
 
bool randomAccess () const
 
void readAndMergeLumi (LuminosityBlockPrincipal &lbp)
 Read next luminosity block (same as a prior lumi) More...
 
void readAndMergeRun (RunPrincipal &rp)
 Read next run (same as a prior run) More...
 
bool readEvent (EventPrincipal &ep, EventID const &, StreamContext &)
 Read a specific event. More...
 
void readEvent (EventPrincipal &ep, StreamContext &)
 Read next event. More...
 
std::shared_ptr< FileBlock > readFile ()
 Read next file. More...
 
void readLuminosityBlock (LuminosityBlockPrincipal &lumiPrincipal, HistoryAppender &historyAppender)
 Read next luminosity block (new lumi) More...
 
std::shared_ptr< LuminosityBlockAuxiliary > readLuminosityBlockAuxiliary ()
 Read next luminosity block Auxilary. More...
 
void readProcessBlock (ProcessBlockPrincipal &)
 Read next process block. More...
 
void readRun (RunPrincipal &runPrincipal, HistoryAppender &historyAppender)
 Read next run (new run) More...
 
std::shared_ptr< RunAuxiliary > readRunAuxiliary ()
 Read next run Auxiliary. More...
 
ProcessHistoryID const & reducedProcessHistoryID () const
 
virtual void registerProducts ()
 Register any produced products. More...
 
int remainingEvents () const
 
int remainingLuminosityBlocks () const
 
void repeat ()
 Reset the remaining number of events/lumis to the maximum number. More...
 
std::pair< SharedResourcesAcquirer *, std::recursive_mutex * > resourceSharedWithDelayedReader ()
 Returns nullptr if no resource shared between the Source and a DelayedReader. More...
 
ProcessingController::ReverseState reverseState () const
 
void rewind ()
 Begin again at the first event. More...
 
RunNumber_t run () const
 Accessor for current run number. More...
 
std::shared_ptr< RunAuxiliary > runAuxiliary () const
 Called by the framework to merge or insert run in principal cache. More...
 
void setLuminosityBlockNumber_t (LuminosityBlockNumber_t lb)
 Set the luminosity block ID. More...
 
void setRunNumber (RunNumber_t r)
 Set the run number. More...
 
void skipEvents (int offset)
 
std::shared_ptr< ThinnedAssociationsHelper > & thinnedAssociationsHelper ()
 
std::shared_ptr< ThinnedAssociationsHelper const > thinnedAssociationsHelper () const
 Accessors for thinnedAssociationsHelper. More...
 
Timestamp const & timestamp () const
 Accessor for the current time, as seen by the input source. More...
 
virtual ~InputSource () noexcept(false)
 Destructor. More...
 

Static Public Member Functions

static void fillDescriptions (edm::ConfigurationDescriptions &descriptions)
 
- Static Public Member Functions inherited from edm::RawInputSource
static void fillDescription (ParameterSetDescription &description)
 
- Static Public Member Functions inherited from edm::InputSource
static const std::string & baseType ()
 
static void fillDescription (ParameterSetDescription &desc)
 
static void fillDescriptions (ConfigurationDescriptions &descriptions)
 
static void prevalidate (ConfigurationDescriptions &)
 

Protected Member Functions

Next checkNext () override
 
void read (edm::EventPrincipal &eventPrincipal) override
 
void setMonState (evf::FastMonState::InputState state)
 
void setMonStateSup (evf::FastMonState::InputState state)
 
- Protected Member Functions inherited from edm::RawInputSource
void makeEvent (EventPrincipal &eventPrincipal, EventAuxiliary const &eventAuxiliary)
 
void setInputFileTransitionsEachEvent ()
 
- Protected Member Functions inherited from edm::InputSource
virtual void beginJob ()
 Begin protected makes it easier to do template programming. More...
 
void decreaseRemainingEventsBy (int iSkipped)
 
bool eventCached () const
 
bool newLumi () const
 
bool newRun () const
 
ProcessHistoryRegistry & processHistoryRegistryForUpdate ()
 
ProductRegistry & productRegistryUpdate ()
 
void reset () const
 
void resetEventCached ()
 
void resetLuminosityBlockAuxiliary (bool isNewLumi=true) const
 
void resetNewLumi ()
 
void resetNewRun ()
 
void resetRunAuxiliary (bool isNewRun=true) const
 
void setEventCached ()
 Called by the framework to merge or ached() const {return eventCached_;}. More...
 
void setLuminosityBlockAuxiliary (LuminosityBlockAuxiliary *lbp)
 
void setNewLumi ()
 
void setNewRun ()
 
void setRunAuxiliary (RunAuxiliary *rp)
 
void setTimestamp (Timestamp const &theTime)
 To set the current time, as seen by the input source. More...
 
virtual void skip (int offset)
 
ItemType state () const
 

Private Types

typedef std::pair< InputFile *, InputChunk * > ReaderInfo
 

Private Member Functions

bool exceptionState ()
 
edm::Timestamp fillFEDRawDataCollection (FEDRawDataCollection &rawData, bool &tcdsInRange)
 
evf::EvFDaqDirector::FileStatus getFile (unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint64_t &lockWaitTime)
 
evf::EvFDaqDirector::FileStatus getNextEvent ()
 
long initFileList ()
 
void maybeOpenNewLumiSection (const uint32_t lumiSection)
 
evf::EvFDaqDirector::FileStatus nextEvent ()
 
void readNextChunkIntoBuffer (InputFile *file)
 
void readSupervisor ()
 
void readWorker (unsigned int tid)
 
void reportEventsThisLumiInSource (unsigned int lumi, unsigned int events)
 
void rewind_ () override
 
void threadError ()
 

Private Attributes

const bool alwaysStartFromFirstLS_
 
uint32_t bufferInputRead_ = 0
 
unsigned int checkEvery_ = 10
 
bool chunkIsFree_ = false
 
std::unique_ptr< InputFile > currentFile_
 
int currentFileIndex_ = -1
 
unsigned int currentLumiSection_
 
std::vector< std::condition_variable * > cvReader_
 
std::condition_variable cvWakeup_
 
evf::EvFDaqDirector * daqDirector_ = nullptr
 
const edm::DaqProvenanceHelper daqProvenanceHelper_
 
std::string defPath_
 
uint16_t detectedFRDversion_ = 0
 
std::unique_ptr< FRDEventMsgView > event_
 
unsigned int eventChunkBlock_
 
unsigned int eventChunkSize_
 
edm::EventID eventID_
 
uint32_t eventRunNumber_ = 0
 
unsigned int eventsThisLumi_
 
unsigned long eventsThisRun_ = 0
 
std::mutex fileDeleteLock_
 
int fileDescriptor_ = -1
 
unsigned int fileListIndex_ = 0
 
const bool fileListLoopMode_
 
const bool fileListMode_
 
std::vector< std::string > fileNames_
 
std::list< std::pair< int, std::string > > fileNamesToDelete_
 
tbb::concurrent_queue< std::unique_ptr< InputFile > > fileQueue_
 
std::list< std::pair< int, std::unique_ptr< InputFile > > > filesToDelete_
 
evf::FastMonitoringService * fms_ = nullptr
 
tbb::concurrent_queue< InputChunk * > freeChunks_
 
std::string fuOutputDir_
 
const bool getLSFromFilename_
 
uint32_t GTPEventID_ = 0
 
uint32_t L1EventID_ = 0
 
unsigned int loopModeIterationInc_ = 0
 
unsigned int maxBufferedFiles_
 
uint16_t MAXTCDSuTCAFEDID_ = FEDNumbering::MAXTCDSuTCAFEDID
 
uint16_t MINTCDSuTCAFEDID_ = FEDNumbering::MINTCDSuTCAFEDID
 
std::mutex monlock_
 
std::mutex mReader_
 
std::mutex mWakeup_
 
unsigned int nStreams_ = 0
 
unsigned int numBuffers_
 
unsigned int numConcurrentReads_
 
edm::ProcessHistoryID processHistoryID_
 
std::atomic< bool > quit_threads_
 
unsigned int readBlocks_
 
std::atomic< unsigned int > readingFilesCount_
 
std::unique_ptr< std::thread > readSupervisorThread_
 
edm::RunNumber_t runNumber_
 
bool setExceptionState_ = false
 
bool singleBufferMode_
 
std::map< unsigned int, unsigned int > sourceEventsReport_
 
bool startedSupervisorThread_ = false
 
std::condition_variable startupCv_
 
std::mutex startupLock_
 
std::vector< int > streamFileTracker_
 
unsigned char * tcds_pointer_
 
const std::vector< unsigned int > testTCDSFEDRange_
 
std::vector< unsigned int > thread_quit_signal
 
std::atomic< bool > threadInit_
 
std::vector< unsigned int > tid_active_
 
bool useFileBroker_
 
const bool useL1EventID_
 
const bool verifyChecksum_
 
std::vector< ReaderInfo > workerJob_
 
tbb::concurrent_queue< unsigned int > workerPool_
 
std::vector< std::thread * > workerThreads_
 

Friends

struct InputChunk
 
struct InputFile
 

Additional Inherited Members

- Public Types inherited from edm::RawInputSource
enum  Next { Next::kEvent, Next::kFile, Next::kStop }
 
- Public Types inherited from edm::InputSource
enum  ItemType {
  IsInvalid, IsStop, IsFile, IsRun,
  IsLumi, IsEvent, IsRepeat, IsSynchronize
}
 
enum  ProcessingMode { Runs, RunsAndLumis, RunsLumisAndEvents }
 
- Public Attributes inherited from edm::InputSource
signalslot::Signal< void(StreamContext const &, ModuleCallingContext const &)> postEventReadFromSourceSignal_
 
signalslot::Signal< void(StreamContext const &, ModuleCallingContext const &)> preEventReadFromSourceSignal_
 

Detailed Description

Definition at line 40 of file FedRawDataInputSource.h.

Member Typedef Documentation

◆ ReaderInfo

typedef std::pair<InputFile*, InputChunk*> FedRawDataInputSource::ReaderInfo
private

Definition at line 138 of file FedRawDataInputSource.h.

Constructor & Destructor Documentation

◆ FedRawDataInputSource()

FedRawDataInputSource::FedRawDataInputSource ( edm::ParameterSet const &  pset,
edm::InputSourceDescription const &  desc 
)
explicit

Definition at line 53 of file FedRawDataInputSource.cc.

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  }
86  MINTCDSuTCAFEDID_ = testTCDSFEDRange_[0];
87  MAXTCDSuTCAFEDID_ = testTCDSFEDRange_[1];
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 
102  processHistoryID_ = daqProvenanceHelper_.daqInit(productRegistryUpdate(), processHistoryRegistryForUpdate());
103  setNewRun();
104  //todo:autodetect from file name (assert if names differ)
105  setRunAuxiliary(new edm::RunAuxiliary(runNumber_, edm::Timestamp::beginOfTime(), edm::Timestamp::invalidTimestamp()));
106 
107  //make sure that chunk size is N * block size
108  assert(eventChunkSize_ >= eventChunkBlock_);
109  readBlocks_ = eventChunkSize_ / eventChunkBlock_;
110  if (readBlocks_ * eventChunkBlock_ != eventChunkSize_)
111  eventChunkSize_ = readBlocks_ * eventChunkBlock_;
112 
113  if (!numBuffers_)
114  throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource")
115  << "no reading enabled with numBuffers parameter 0";
116 
117  numConcurrentReads_ = numBuffers_ - 1;
118  singleBufferMode_ = !(numBuffers_ > 1);
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 
138  daqDirector_ = edm::Service<evf::EvFDaqDirector>().operator->();
139  if (!daqDirector_)
140  cms::Exception("FedRawDataInputSource") << "EvFDaqDirector not found";
141 
142  useFileBroker_ = daqDirector_->useFileBroker();
143  if (useFileBroker_)
144  edm::LogInfo("FedRawDataInputSource") << "EvFDaqDirector/Source configured to use file service";
145  //set DaqDirector to delete files in preGlobalEndLumi callback
146  daqDirector_->setDeleteTracking(&fileDeleteLock_, &filesToDelete_);
147  if (fms_) {
148  daqDirector_->setFMS(fms_);
149  fms_->setInputSource(this);
150  fms_->setInState(inInit);
151  fms_->setInStateSup(inInit);
152  }
153  //should delete chunks when run stops
154  for (unsigned int i = 0; i < numBuffers_; i++) {
155  freeChunks_.push(new InputChunk(i, eventChunkSize_));
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 }

References cms::cuda::assert(), edm::Timestamp::beginOfTime(), crc32c_hw_test(), cvReader_, daqDirector_, edm::DaqProvenanceHelper::daqInit(), daqProvenanceHelper_, eventChunkBlock_, eventChunkSize_, Exception, fileDeleteLock_, fileListLoopMode_, fileListMode_, filesToDelete_, fms_, freeChunks_, mps_fire::i, evf::FastMonState::inInit, initFileList(), InputChunk, edm::Timestamp::invalidTimestamp(), MAXTCDSuTCAFEDID_, MINTCDSuTCAFEDID_, numBuffers_, numConcurrentReads_, Utilities::operator, processHistoryID_, edm::InputSource::processHistoryRegistryForUpdate(), edm::InputSource::productRegistryUpdate(), quit_threads_, readBlocks_, readingFilesCount_, readWorker(), edm::InputSource::runAuxiliary(), runNumber_, evf::EvFDaqDirector::setDeleteTracking(), evf::EvFDaqDirector::setFMS(), evf::FastMonitoringService::setInputSource(), evf::FastMonitoringService::setInState(), evf::FastMonitoringService::setInStateSup(), edm::InputSource::setNewRun(), edm::InputSource::setRunAuxiliary(), singleBufferMode_, startupCv_, startupLock_, testTCDSFEDRange_, thread_quit_signal, threadInit_, tid_active_, evf::EvFDaqDirector::useFileBroker(), useFileBroker_, workerJob_, and workerThreads_.

◆ ~FedRawDataInputSource()

FedRawDataInputSource::~FedRawDataInputSource ( )
override

Definition at line 175 of file FedRawDataInputSource.cc.

175  {
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 
182  if (startedSupervisorThread_) {
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 }

References cvReader_, filesToDelete_, mps_fire::i, mReader_, numConcurrentReads_, quit_threads_, readSupervisorThread_, startedSupervisorThread_, thread_quit_signal, and workerThreads_.

Member Function Documentation

◆ checkNext()

edm::RawInputSource::Next FedRawDataInputSource::checkNext ( )
overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 231 of file FedRawDataInputSource.cc.

231  {
232  if (!startedSupervisorThread_) {
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);
237  startedSupervisorThread_ = true;
238  startupCv_.wait(lk);
239  }
240  //signal hltd to start event accounting
241  if (!currentLumiSection_)
242  daqDirector_->createProcessingNotificationMaybe();
243  setMonState(inWaitInput);
244  switch (nextEvent()) {
245  case evf::EvFDaqDirector::runEnded: {
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) {
251  const std::string fuEoLS = daqDirector_->getEoLSFilePathOnFU(currentLumiSection_);
252  bool found = (stat(fuEoLS.c_str(), &buf) == 0);
253  if (!found) {
254  daqDirector_->lockFULocal2();
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);
258  daqDirector_->unlockFULocal2();
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  }
270  reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
271  eventsThisLumi_ = 0;
272  resetLuminosityBlockAuxiliary();
273  edm::LogInfo("FedRawDataInputSource") << "----------------RUN ENDED----------------";
274  return Next::kStop;
275  }
276  case evf::EvFDaqDirector::noFile: {
277  //this is not reachable
278  return Next::kEvent;
279  }
280  case evf::EvFDaqDirector::newLumi: {
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_) {
288  reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
289  eventsThisLumi_ = 0;
290  maybeOpenNewLumiSection(event_->lumi());
291  }
292  }
293  if (fileListMode_ || fileListLoopMode_)
294  eventRunNumber_ = runNumber_;
295  else
296  eventRunNumber_ = event_->run();
297  L1EventID_ = event_->event();
298 
299  setEventCached();
300 
301  return Next::kEvent;
302  }
303  }
304 }

References visDQMUpload::buf, evf::EvFDaqDirector::createProcessingNotificationMaybe(), currentLumiSection_, daqDirector_, event_, eventRunNumber_, eventsThisLumi_, fileListLoopMode_, fileListMode_, newFWLiteAna::found, evf::EvFDaqDirector::getEoLSFilePathOnBU(), evf::EvFDaqDirector::getEoLSFilePathOnFU(), evf::EvFDaqDirector::getEoRFilePathOnFU(), getLSFromFilename_, evf::FastMonState::inWaitInput, edm::RawInputSource::kEvent, edm::RawInputSource::kStop, L1EventID_, evf::EvFDaqDirector::lockFULocal2(), maybeOpenNewLumiSection(), evf::EvFDaqDirector::newLumi, nextEvent(), evf::EvFDaqDirector::noFile, readSupervisor(), readSupervisorThread_, reportEventsThisLumiInSource(), edm::InputSource::resetLuminosityBlockAuxiliary(), evf::EvFDaqDirector::runEnded, runNumber_, edm::InputSource::setEventCached(), setMonState(), startedSupervisorThread_, startupCv_, startupLock_, edm_modernize_messagelogger::stat, AlCaHLTBitMon_QueryRunRegistry::string, evf::EvFDaqDirector::unlockFULocal2(), and useFileBroker_.

◆ exceptionState()

bool FedRawDataInputSource::exceptionState ( )
inlineprivate

Definition at line 68 of file FedRawDataInputSource.h.

68 { return setExceptionState_; }

References setExceptionState_.

Referenced by InputFile::advance().

◆ fillDescriptions()

void FedRawDataInputSource::fillDescriptions ( edm::ConfigurationDescriptions &  descriptions)
static

Definition at line 206 of file FedRawDataInputSource.cc.

206  {
207  edm::ParameterSetDescription desc;
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 }

References edm::ConfigurationDescriptions::add(), and submitPVResolutionJobs::desc.

◆ fillFEDRawDataCollection()

edm::Timestamp FedRawDataInputSource::fillFEDRawDataCollection ( FEDRawDataCollection &  rawData,
bool &  tcdsInRange 
)
private

Definition at line 689 of file FedRawDataInputSource.cc.

689  {
690  edm::TimeValue_t time;
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();
712  if (fedId > FEDNumbering::MAXFEDID) {
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;
718  if (fedId >= FEDNumbering::MINTCDSuTCAFEDID && fedId <= FEDNumbering::MAXTCDSuTCAFEDID) {
719  tcdsInRange = true;
720  }
721  } else
722  throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection")
723  << "Second TCDS FED ID " << fedId << " found. First ID: " << selectedTCDSFed;
724  }
725  if (fedId == FEDNumbering::MINTriggerGTPFEDID) {
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
736  if (fedId == FEDNumbering::MINTriggerEGTPFEDID && GTPEventID_ == 0) {
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 }

References cms::cuda::assert(), FEDRawData::data(), event_, evf::evtn::evm_board_sense(), Exception, l1tstage2_dqm_sourceclient-live_cfg::fedId, FEDTrailer::fragmentLength(), evf::evtn::get(), evf::evtn::getgpshigh(), evf::evtn::getgpslow(), evf::evtn::gtpe_board_sense(), evf::evtn::gtpe_get(), GTPEventID_, FEDHeader::length, FEDTrailer::length, FEDNumbering::MAXFEDID, FEDNumbering::MAXTCDSuTCAFEDID, MAXTCDSuTCAFEDID_, FEDNumbering::MINTCDSuTCAFEDID, MINTCDSuTCAFEDID_, FEDNumbering::MINTriggerEGTPFEDID, FEDNumbering::MINTriggerGTPFEDID, l1tstage2_dqm_sourceclient-live_cfg::rawData, FEDRawData::resize(), FEDHeader::sourceID(), tcds_pointer_, and protons_cff::time.

Referenced by read().

◆ getEventReport()

std::pair< bool, unsigned int > FedRawDataInputSource::getEventReport ( unsigned int  lumi,
bool  erase 
)

Definition at line 1516 of file FedRawDataInputSource.cc.

1516  {
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 }

References CommonMethods::lock(), monlock_, runTheMatrix::ret, and sourceEventsReport_.

Referenced by evf::FastMonitoringService::preGlobalEndLumi().

◆ getFile()

evf::EvFDaqDirector::FileStatus FedRawDataInputSource::getFile ( unsigned int &  ls,
std::string &  nextFile,
uint32_t &  fsize,
uint64_t &  lockWaitTime 
)
private

Definition at line 1563 of file FedRawDataInputSource.cc.

1566  {
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_++;
1588  return evf::EvFDaqDirector::newFile;
1589  } else {
1590  if (!fileListLoopMode_)
1591  return evf::EvFDaqDirector::runEnded;
1592  else {
1593  //loop through files until interrupted
1594  loopModeIterationInc_++;
1595  fileListIndex_ = 0;
1596  return getFile(ls, nextFile, fsize, lockWaitTime);
1597  }
1598  }
1599 }

References fileListIndex_, fileListLoopMode_, MillePedeFileConverter_cfg::fileName, fileNames_, loopModeIterationInc_, eostools::ls(), evf::EvFDaqDirector::newFile, castor_dqm_sourceclient_file_cfg::path, evf::EvFDaqDirector::runEnded, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by readSupervisor().

◆ getNextEvent()

evf::EvFDaqDirector::FileStatus FedRawDataInputSource::getNextEvent ( )
inlineprivate

Definition at line 352 of file FedRawDataInputSource.cc.

352  {
353  if (setExceptionState_)
354  threadError();
355  if (!currentFile_.get()) {
356  evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
357  setMonState(inWaitInput);
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())
362  return evf::EvFDaqDirector::noFile;
363  }
364  status = currentFile_->status_;
365  if (status == evf::EvFDaqDirector::runEnded) {
366  setMonState(inRunEnd);
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) {
373  setMonState(inNewLumi);
374  if (getLSFromFilename_) {
375  if (currentFile_->lumi_ > currentLumiSection_) {
376  reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
377  eventsThisLumi_ = 0;
378  maybeOpenNewLumiSection(currentFile_->lumi_);
379  }
380  } else { //let this be picked up from next event
381  status = evf::EvFDaqDirector::noFile;
382  }
383  currentFile_.reset();
384  return status;
385  } else if (status == evf::EvFDaqDirector::newFile) {
386  currentFileIndex_++;
387  } else
388  assert(false);
389  }
390  setMonState(inProcessingFile);
391 
392  //file is empty
393  if (!currentFile_->fileSize_) {
394  readingFilesCount_--;
395  //try to open new lumi
396  assert(currentFile_->nChunks_ == 0);
397  if (getLSFromFilename_)
398  if (currentFile_->lumi_ > currentLumiSection_) {
399  reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
400  eventsThisLumi_ = 0;
401  maybeOpenNewLumiSection(currentFile_->lumi_);
402  }
403  //immediately delete empty file
404  currentFile_.reset();
405  return evf::EvFDaqDirector::noFile;
406  }
407 
408  //file is finished
409  if (currentFile_->bufferPosition_ == currentFile_->fileSize_) {
410  readingFilesCount_--;
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;
424  if (!daqDirector_->isSingleStreamThread() && !fileListMode_) {
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  }
432  return evf::EvFDaqDirector::noFile;
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_) {
446  reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
447  eventsThisLumi_ = 0;
448  maybeOpenNewLumiSection(currentFile_->lumi_);
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
464  setMonState(inWaitChunk);
465  while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
466  usleep(10000);
467  if (currentFile_->parent_->exceptionState() || setExceptionState_)
468  currentFile_->parent_->threadError();
469  }
470  setMonState(inChunkReceived);
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
475  if (!bufferInputRead_ || bufferInputRead_ < FRDHeaderVersionSize[detectedFRDversion_] ||
476  eventChunkSize_ - currentFile_->chunkPosition_ < FRDHeaderVersionSize[detectedFRDversion_]) {
477  readNextChunkIntoBuffer(currentFile_.get());
478 
479  if (detectedFRDversion_ == 0) {
480  detectedFRDversion_ = *((uint16_t*)dataPosition);
481  if (detectedFRDversion_ > FRDHeaderMaxVersion)
482  throw cms::Exception("FedRawDataInputSource::getNextEvent")
483  << "Unknown FRD version -: " << detectedFRDversion_;
484  assert(detectedFRDversion_ >= 1);
485  }
486 
487  //recalculate chunk position
488  dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
489  if (bufferInputRead_ < FRDHeaderVersionSize[detectedFRDversion_]) {
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) {
510  readNextChunkIntoBuffer(currentFile_.get());
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
523  setMonState(inWaitChunk);
524  while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
525  usleep(10000);
526  if (setExceptionState_)
527  threadError();
528  }
529  setMonState(inChunkReceived);
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
561  currentFile_->rewindChunk(FRDHeaderVersionSize[detectedFRDversion_]);
562  //copy event to a chunk start and move pointers
563 
564  setMonState(inWaitChunk);
565 
566  chunkEnd = currentFile_->advance(dataPosition, FRDHeaderVersionSize[detectedFRDversion_] + msgSize);
567 
568  setMonState(inChunkReceived);
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
582  setMonState(inChecksumEvent);
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_)
589  fms_->setExceptionDetected(currentLumiSection_);
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_)
600  fms_->setExceptionDetected(currentLumiSection_);
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  }
606  setMonState(inCachedEvent);
607 
608  currentFile_->nProcessed_++;
609 
610  return evf::EvFDaqDirector::sameFile;
611 }

References cms::cuda::assert(), bufferInputRead_, chunkIsFree_, crc32c(), currentFile_, currentFileIndex_, currentLumiSection_, cvWakeup_, daqDirector_, detectedFRDversion_, event_, eventChunkSize_, eventsThisLumi_, Exception, fileDeleteLock_, fileListMode_, fileQueue_, filesToDelete_, fms_, FRDHeaderMaxVersion, FRDHeaderVersionSize, freeChunks_, getLSFromFilename_, evf::FastMonState::inCachedEvent, evf::FastMonState::inChecksumEvent, evf::FastMonState::inChunkReceived, evf::FastMonState::inNewLumi, evf::FastMonState::inProcessingFile, evf::FastMonState::inRunEnd, evf::FastMonState::inWaitChunk, evf::FastMonState::inWaitInput, evf::EvFDaqDirector::isSingleStreamThread(), dttmaxenums::L, maybeOpenNewLumiSection(), eostools::move(), mWakeup_, evf::EvFDaqDirector::newFile, evf::EvFDaqDirector::newLumi, evf::EvFDaqDirector::noFile, readingFilesCount_, readNextChunkIntoBuffer(), reportEventsThisLumiInSource(), evf::EvFDaqDirector::runAbort, evf::EvFDaqDirector::runEnded, evf::EvFDaqDirector::sameFile, evf::FastMonitoringService::setExceptionDetected(), setExceptionState_, setMonState(), singleBufferMode_, mps_update::status, threadError(), mps_check::timeout, and verifyChecksum_.

Referenced by nextEvent().

◆ initFileList()

long FedRawDataInputSource::initFileList ( )
private

Definition at line 1528 of file FedRawDataInputSource.cc.

1528  {
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
1539  std::filesystem::path fileName = fileNames_[0];
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 }

References a, b, mps_fire::end, MillePedeFileConverter_cfg::fileName, fileNames_, castor_dqm_sourceclient_file_cfg::path, jetUpdater_cfi::sort, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource().

◆ maybeOpenNewLumiSection()

void FedRawDataInputSource::maybeOpenNewLumiSection ( const uint32_t  lumiSection)
private

Definition at line 306 of file FedRawDataInputSource.cc.

306  {
307  if (!luminosityBlockAuxiliary() || luminosityBlockAuxiliary()->luminosityBlock() != lumiSection) {
308  if (!useFileBroker_) {
309  if (currentLumiSection_ > 0) {
310  const std::string fuEoLS = daqDirector_->getEoLSFilePathOnFU(currentLumiSection_);
311  struct stat buf;
312  bool found = (stat(fuEoLS.c_str(), &buf) == 0);
313  if (!found) {
314  daqDirector_->lockFULocal2();
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);
319  daqDirector_->unlockFULocal2();
320  }
321  } else
322  daqDirector_->createBoLSFile(lumiSection, true); //needed for initial lumisection
323  }
324 
325  currentLumiSection_ = lumiSection;
326 
327  resetLuminosityBlockAuxiliary();
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 
333  edm::LuminosityBlockAuxiliary* lumiBlockAuxiliary = new edm::LuminosityBlockAuxiliary(
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 }

References visDQMUpload::buf, evf::EvFDaqDirector::createBoLSFile(), currentLumiSection_, daqDirector_, newFWLiteAna::found, evf::EvFDaqDirector::getEoLSFilePathOnFU(), edm::Timestamp::invalidTimestamp(), evf::EvFDaqDirector::lockFULocal2(), edm::InputSource::luminosityBlock(), edm::InputSource::luminosityBlockAuxiliary(), processHistoryID_, edm::InputSource::resetLuminosityBlockAuxiliary(), edm::InputSource::run(), edm::InputSource::runAuxiliary(), edm::InputSource::setLuminosityBlockAuxiliary(), edm_modernize_messagelogger::stat, AlCaHLTBitMon_QueryRunRegistry::string, evf::EvFDaqDirector::unlockFULocal2(), and useFileBroker_.

Referenced by checkNext(), and getNextEvent().

◆ nextEvent()

evf::EvFDaqDirector::FileStatus FedRawDataInputSource::nextEvent ( )
inlineprivate

Definition at line 343 of file FedRawDataInputSource.cc.

343  {
344  evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
345  while ((status = getNextEvent()) == evf::EvFDaqDirector::noFile) {
346  if (edm::shutdown_flag.load(std::memory_order_relaxed))
347  break;
348  }
349  return status;
350 }

References getNextEvent(), svgfig::load(), evf::EvFDaqDirector::noFile, edm::shutdown_flag, and mps_update::status.

Referenced by checkNext().

◆ read()

void FedRawDataInputSource::read ( edm::EventPrincipal &  eventPrincipal)
overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 613 of file FedRawDataInputSource.cc.

613  {
614  setMonState(inReadEvent);
615  std::unique_ptr<FEDRawDataCollection> rawData(new FEDRawDataCollection);
616  bool tcdsInRange;
617  edm::Timestamp tstamp = fillFEDRawDataCollection(*rawData, tcdsInRange);
618 
619  if (useL1EventID_) {
620  eventID_ = edm::EventID(eventRunNumber_, currentLumiSection_, L1EventID_);
621  edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, event_->isRealData(), edm::EventAuxiliary::PhysicsTrigger);
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  }
629  eventID_ = edm::EventID(eventRunNumber_, currentLumiSection_, GTPEventID_);
630  edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, event_->isRealData(), edm::EventAuxiliary::PhysicsTrigger);
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);
636  edm::EventAuxiliary aux =
637  evf::evtn::makeEventAuxiliary(tcds,
638  eventRunNumber_,
639  currentLumiSection_,
640  event_->isRealData(),
641  static_cast<edm::EventAuxiliary::ExperimentType>(fedHeader.triggerType()),
642  processGUID(),
643  !fileListLoopMode_,
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 
651  eventPrincipal.put(daqProvenanceHelper_.branchDescription(), std::move(edp), daqProvenanceHelper_.dummyProvenance());
652 
653  eventsThisLumi_++;
654  setMonState(inReadCleanup);
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;
685  setMonState(inNoRequest);
686  return;
687 }

References printConversionInfo::aux, edm::DaqProvenanceHelper::branchDescription(), checkEvery_, chunkIsFree_, currentFile_, currentFileIndex_, currentLumiSection_, daqProvenanceHelper_, edm::DaqProvenanceHelper::dummyProvenance(), event_, eventID_, eventRunNumber_, eventsThisLumi_, Exception, fileDeleteLock_, fileListLoopMode_, filesToDelete_, fillFEDRawDataCollection(), freeChunks_, GTPEventID_, mps_fire::i, evf::FastMonState::inNoRequest, evf::FastMonState::inReadCleanup, evf::FastMonState::inReadEvent, L1EventID_, FEDHeader::length, edm::RawInputSource::makeEvent(), evf::evtn::makeEventAuxiliary(), eostools::move(), nStreams_, edm::EventAuxiliary::PhysicsTrigger, edm::InputSource::processGUID(), processHistoryID_, edm::EventPrincipal::put(), l1tstage2_dqm_sourceclient-live_cfg::rawData, setMonState(), streamFileTracker_, edm::EventPrincipal::streamID(), AlCaHLTBitMon_QueryRunRegistry::string, tcds_pointer_, FEDHeader::triggerType(), and useL1EventID_.

Referenced by edmIntegrityCheck.PublishToFileSystem::get(), readNextChunkIntoBuffer(), and readWorker().

◆ readNextChunkIntoBuffer()

void FedRawDataInputSource::readNextChunkIntoBuffer ( InputFile *  file)
private

Definition at line 1436 of file FedRawDataInputSource.cc.

1436  {
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));
1464  bufferInputRead_ += last;
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_);
1473  bufferInputRead_ += last;
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_);
1488  bufferInputRead_ += last;
1489  existingSizeLeft += last;
1490  }
1491  if (leftsize) {
1492  const ssize_t last = ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSizeLeft), leftsize);
1493  bufferInputRead_ += last;
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 }

References bufferInputRead_, eventChunkBlock_, eventChunkSize_, Exception, geometryDiff::file, fileDescriptor_, mps_fire::i, dqmdumpme::last, LogDebug, read(), and readBlocks_.

Referenced by getNextEvent().

◆ readSupervisor()

void FedRawDataInputSource::readSupervisor ( )
private

Definition at line 752 of file FedRawDataInputSource.cc.

752  {
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() ||
771  readingFilesCount_ >= maxBufferedFiles_) {
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;
778  if (readingFilesCount_ >= maxBufferedFiles_)
779  setMonStateSup(inSupFileLimit);
780  else if (freeChunks_.empty()) {
781  if (copy_active)
782  setMonStateSup(inSupWaitFreeChunkCopying);
783  else
784  setMonStateSup(inSupWaitFreeChunk);
785  } else {
786  if (copy_active)
787  setMonStateSup(inSupWaitFreeThreadCopying);
788  else
789  setMonStateSup(inSupWaitFreeThread);
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_) {
817  setMonStateSup(inSupBusy);
818  fms_->startedLookingForFile();
819  }
820 
821  evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
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
830  while (status == evf::EvFDaqDirector::noFile) {
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.;
838  setMonStateSup(inSupLockPolling);
839  if (fileListMode_) {
840  //return LS if LS not set, otherwise return file
841  status = getFile(ls, nextFile, fileSizeIndex, thisLockWaitTimeUs);
842  if (status == evf::EvFDaqDirector::newFile) {
843  if (evf::EvFDaqDirector::parseFRDFileHeader(nextFile,
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_)
861  status = daqDirector_->updateFuLock(
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 
874  setMonStateSup(inSupBusy);
875 
876  //cycle through all remaining LS even if no files get assigned
877  if (currentLumiSection != ls && status == evf::EvFDaqDirector::runEnded)
878  status = evf::EvFDaqDirector::noFile;
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
894  if (status == evf::EvFDaqDirector::runEnded && !fileListMode_ && !useFileBroker_) {
895  setMonStateSup(inRunEnd);
896  usleep(100000);
897  //now all files should have appeared in ramdisk, check again if any raw files were left behind
898  status = daqDirector_->updateFuLock(
899  ls, nextFile, fileSizeIndex, rawHeaderSize, thisLockWaitTimeUs, setExceptionState_);
900  if (currentLumiSection != ls && status == evf::EvFDaqDirector::runEnded)
901  status = evf::EvFDaqDirector::noFile;
902  }
903 
904  if (status == evf::EvFDaqDirector::runEnded) {
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
912  if (status == evf::EvFDaqDirector::runAbort) {
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 = std::min(lsToStart, ls); 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;
969  if (status == evf::EvFDaqDirector::noFile) {
970  setMonStateSup(inSupNoFile);
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
987  if (status == evf::EvFDaqDirector::newFile) {
988  setMonStateSup(inSupNewFile);
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_) {
1010  setMonStateSup(inSupBusy);
1011  fms_->stoppedLookingForFile(ls);
1012  setMonStateSup(inSupNewFile);
1013  }
1014  int eventsInNewFile;
1015  if (fileListMode_) {
1016  if (fileSize == 0)
1017  eventsInNewFile = 0;
1018  else
1019  eventsInNewFile = -1;
1020  } else {
1021  std::string empty;
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);
1030  daqDirector_->unlockFULocal();
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));
1056  readingFilesCount_++;
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)
1067  setMonStateSup(inSupNewFileWaitThreadCopying);
1068  else
1069  setMonStateSup(inSupNewFileWaitThread);
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)
1087  setMonStateSup(inSupNewFileWaitChunkCopying);
1088  else
1089  setMonStateSup(inSupNewFileWaitChunk);
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;
1109  setMonStateSup(inSupNewFile);
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));
1143  readingFilesCount_++;
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;
1177  readingFilesCount_++;
1178  fileQueue_.push(std::move(newInputFile));
1179  cvWakeup_.notify_one();
1180  }
1181  }
1182  }
1183  setMonStateSup(inRunEnd);
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 }

References alwaysStartFromFirstLS_, cms::cuda::assert(), cvReader_, cvWakeup_, daqDirector_, relativeConstraints::empty, eventChunkSize_, fileListMode_, fileQueue_, fms_, freeChunks_, getFile(), getLSFromFilename_, evf::EvFDaqDirector::getLumisectionToStart(), evf::EvFDaqDirector::getNextFromFileBroker(), evf::EvFDaqDirector::grabNextJsonFileAndUnlock(), evf::EvFDaqDirector::grabNextJsonFromRaw(), mps_fire::i, dqmiodatasetharvest::inf, InputFile, evf::FastMonState::inRunEnd, evf::FastMonState::inSupBusy, evf::FastMonState::inSupFileLimit, evf::FastMonState::inSupLockPolling, evf::FastMonState::inSupNewFile, evf::FastMonState::inSupNewFileWaitChunk, evf::FastMonState::inSupNewFileWaitChunkCopying, evf::FastMonState::inSupNewFileWaitThread, evf::FastMonState::inSupNewFileWaitThreadCopying, evf::FastMonState::inSupNoFile, evf::FastMonState::inSupWaitFreeChunk, evf::FastMonState::inSupWaitFreeChunkCopying, evf::FastMonState::inSupWaitFreeThread, evf::FastMonState::inSupWaitFreeThreadCopying, createfilelist::int, dqmiolumiharvest::j, LogDebug, eostools::ls(), maxBufferedFiles_, min(), eostools::move(), mReader_, mWakeup_, evf::EvFDaqDirector::newFile, evf::EvFDaqDirector::newLumi, evf::EvFDaqDirector::noFile, evf::EvFDaqDirector::parseFRDFileHeader(), castor_dqm_sourceclient_file_cfg::path, funct::pow(), quit_threads_, InputChunk::readComplete_, readingFilesCount_, evf::FastMonitoringService::reportLockWait(), InputChunk::reset(), evf::EvFDaqDirector::runAbort, evf::EvFDaqDirector::runEnded, setExceptionState_, setMonStateSup(), edm::shutdown_flag, singleBufferMode_, evf::FastMonitoringService::startedLookingForFile(), startupCv_, startupLock_, edm_modernize_messagelogger::stat, mps_update::status, evf::FastMonitoringService::stoppedLookingForFile(), AlCaHLTBitMon_QueryRunRegistry::string, thread_quit_signal, tid_active_, mps_check::timeout, evf::EvFDaqDirector::unlockFULocal(), evf::EvFDaqDirector::updateFuLock(), useFileBroker_, workerJob_, workerPool_, and workerThreads_.

Referenced by checkNext().

◆ readWorker()

void FedRawDataInputSource::readWorker ( unsigned int  tid)
private

Definition at line 1202 of file FedRawDataInputSource.cc.

1202  {
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;
1285  auto start = std::chrono::high_resolution_clock::now();
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 
1317  auto end = std::chrono::high_resolution_clock::now();
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  }
1337  assert(detectedFRDversion_ <= FRDHeaderMaxVersion);
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 }

References cms::cuda::assert(), InputChunk::buf_, cvReader_, detectedFRDversion_, change_name::diff, mps_fire::end, eventChunkBlock_, geometryDiff::file, InputChunk::fileIndex_, first, FRDHeaderMaxVersion, mps_fire::i, if(), dqmdumpme::last, LogDebug, min(), mReader_, submitPVValidationJobs::now, numConcurrentReads_, InputChunk::offset_, read(), readBlocks_, InputChunk::readComplete_, edm::second(), setExceptionState_, runEdmFileComparison::skipped, command_line::start, startupCv_, startupLock_, thread_quit_signal, threadInit_, tid_active_, InputChunk::usedSize_, workerJob_, and workerPool_.

Referenced by FedRawDataInputSource().

◆ reportEventsThisLumiInSource()

void FedRawDataInputSource::reportEventsThisLumiInSource ( unsigned int  lumi,
unsigned int  events 
)
private

Definition at line 1507 of file FedRawDataInputSource.cc.

1507  {
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
1513  sourceEventsReport_[lumi] = events;
1514 }

References patZpeak::events, CommonMethods::lock(), monlock_, and sourceEventsReport_.

Referenced by checkNext(), and getNextEvent().

◆ rewind_()

void FedRawDataInputSource::rewind_ ( )
overrideprivatevirtual

Reimplemented from edm::InputSource.

Definition at line 750 of file FedRawDataInputSource.cc.

750 {}

◆ setMonState()

void FedRawDataInputSource::setMonState ( evf::FastMonState::InputState  state)
inlineprotected

Definition at line 1349 of file FedRawDataInputSource.cc.

1349  {
1350  if (fms_)
1351  fms_->setInState(state);
1352 }

References fms_, evf::FastMonitoringService::setInState(), and edm::InputSource::state().

Referenced by InputFile::advance(), checkNext(), getNextEvent(), and read().

◆ setMonStateSup()

void FedRawDataInputSource::setMonStateSup ( evf::FastMonState::InputState  state)
inlineprotected

Definition at line 1354 of file FedRawDataInputSource.cc.

1354  {
1355  if (fms_)
1356  fms_->setInStateSup(state);
1357 }

References fms_, evf::FastMonitoringService::setInStateSup(), and edm::InputSource::state().

Referenced by readSupervisor().

◆ threadError()

void FedRawDataInputSource::threadError ( )
private

Definition at line 1344 of file FedRawDataInputSource.cc.

1344  {
1345  quit_threads_ = true;
1346  throw cms::Exception("FedRawDataInputSource:threadError") << " file reader thread error ";
1347 }

References Exception, and quit_threads_.

Referenced by InputFile::advance(), and getNextEvent().

Friends And Related Function Documentation

◆ InputChunk

friend struct InputChunk
friend

Definition at line 42 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

◆ InputFile

friend struct InputFile
friend

Definition at line 41 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

Member Data Documentation

◆ alwaysStartFromFirstLS_

const bool FedRawDataInputSource::alwaysStartFromFirstLS_
private

Definition at line 98 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

◆ bufferInputRead_

uint32_t FedRawDataInputSource::bufferInputRead_ = 0
private

Definition at line 179 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readNextChunkIntoBuffer().

◆ checkEvery_

unsigned int FedRawDataInputSource::checkEvery_ = 10
private

Definition at line 170 of file FedRawDataInputSource.h.

Referenced by read().

◆ chunkIsFree_

bool FedRawDataInputSource::chunkIsFree_ = false
private

Definition at line 142 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

◆ currentFile_

std::unique_ptr<InputFile> FedRawDataInputSource::currentFile_
private

Definition at line 141 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

◆ currentFileIndex_

int FedRawDataInputSource::currentFileIndex_ = -1
private

Definition at line 164 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

◆ currentLumiSection_

unsigned int FedRawDataInputSource::currentLumiSection_
private

Definition at line 121 of file FedRawDataInputSource.h.

Referenced by checkNext(), getNextEvent(), maybeOpenNewLumiSection(), and read().

◆ cvReader_

std::vector<std::condition_variable*> FedRawDataInputSource::cvReader_
private

Definition at line 155 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readSupervisor(), readWorker(), and ~FedRawDataInputSource().

◆ cvWakeup_

std::condition_variable FedRawDataInputSource::cvWakeup_
private

Definition at line 174 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

◆ daqDirector_

evf::EvFDaqDirector* FedRawDataInputSource::daqDirector_ = nullptr
private

Definition at line 84 of file FedRawDataInputSource.h.

Referenced by checkNext(), FedRawDataInputSource(), getNextEvent(), maybeOpenNewLumiSection(), and readSupervisor().

◆ daqProvenanceHelper_

const edm::DaqProvenanceHelper FedRawDataInputSource::daqProvenanceHelper_
private

Definition at line 114 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and read().

◆ defPath_

std::string FedRawDataInputSource::defPath_
private

Definition at line 86 of file FedRawDataInputSource.h.

◆ detectedFRDversion_

uint16_t FedRawDataInputSource::detectedFRDversion_ = 0
private

Definition at line 140 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readWorker().

◆ event_

std::unique_ptr<FRDEventMsgView> FedRawDataInputSource::event_
private

Definition at line 116 of file FedRawDataInputSource.h.

Referenced by checkNext(), fillFEDRawDataCollection(), getNextEvent(), and read().

◆ eventChunkBlock_

unsigned int FedRawDataInputSource::eventChunkBlock_
private

Definition at line 89 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readNextChunkIntoBuffer(), and readWorker().

◆ eventChunkSize_

unsigned int FedRawDataInputSource::eventChunkSize_
private

Definition at line 88 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), getNextEvent(), readNextChunkIntoBuffer(), and readSupervisor().

◆ eventID_

edm::EventID FedRawDataInputSource::eventID_
private

Definition at line 118 of file FedRawDataInputSource.h.

Referenced by read().

◆ eventRunNumber_

uint32_t FedRawDataInputSource::eventRunNumber_ = 0
private

Definition at line 122 of file FedRawDataInputSource.h.

Referenced by checkNext(), and read().

◆ eventsThisLumi_

unsigned int FedRawDataInputSource::eventsThisLumi_
private

Definition at line 126 of file FedRawDataInputSource.h.

Referenced by checkNext(), getNextEvent(), and read().

◆ eventsThisRun_

unsigned long FedRawDataInputSource::eventsThisRun_ = 0
private

Definition at line 127 of file FedRawDataInputSource.h.

◆ fileDeleteLock_

std::mutex FedRawDataInputSource::fileDeleteLock_
private

Definition at line 167 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), getNextEvent(), and read().

◆ fileDescriptor_

int FedRawDataInputSource::fileDescriptor_ = -1
private

Definition at line 178 of file FedRawDataInputSource.h.

Referenced by readNextChunkIntoBuffer().

◆ fileListIndex_

unsigned int FedRawDataInputSource::fileListIndex_ = 0
private

Definition at line 107 of file FedRawDataInputSource.h.

Referenced by getFile().

◆ fileListLoopMode_

const bool FedRawDataInputSource::fileListLoopMode_
private

Definition at line 108 of file FedRawDataInputSource.h.

Referenced by checkNext(), FedRawDataInputSource(), getFile(), and read().

◆ fileListMode_

const bool FedRawDataInputSource::fileListMode_
private

Definition at line 106 of file FedRawDataInputSource.h.

Referenced by checkNext(), FedRawDataInputSource(), getNextEvent(), and readSupervisor().

◆ fileNames_

std::vector<std::string> FedRawDataInputSource::fileNames_
private

Definition at line 102 of file FedRawDataInputSource.h.

Referenced by getFile(), and initFileList().

◆ fileNamesToDelete_

std::list<std::pair<int, std::string> > FedRawDataInputSource::fileNamesToDelete_
private

Definition at line 166 of file FedRawDataInputSource.h.

◆ fileQueue_

tbb::concurrent_queue<std::unique_ptr<InputFile> > FedRawDataInputSource::fileQueue_
private

Definition at line 152 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

◆ filesToDelete_

std::list<std::pair<int, std::unique_ptr<InputFile> > > FedRawDataInputSource::filesToDelete_
private

Definition at line 165 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), getNextEvent(), read(), and ~FedRawDataInputSource().

◆ fms_

evf::FastMonitoringService* FedRawDataInputSource::fms_ = nullptr
private

Definition at line 83 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), getNextEvent(), readSupervisor(), setMonState(), and setMonStateSup().

◆ freeChunks_

tbb::concurrent_queue<InputChunk*> FedRawDataInputSource::freeChunks_
private

Definition at line 151 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), getNextEvent(), read(), and readSupervisor().

◆ fuOutputDir_

std::string FedRawDataInputSource::fuOutputDir_
private

Definition at line 112 of file FedRawDataInputSource.h.

◆ getLSFromFilename_

const bool FedRawDataInputSource::getLSFromFilename_
private

Definition at line 97 of file FedRawDataInputSource.h.

Referenced by checkNext(), getNextEvent(), and readSupervisor().

◆ GTPEventID_

uint32_t FedRawDataInputSource::GTPEventID_ = 0
private

Definition at line 123 of file FedRawDataInputSource.h.

Referenced by fillFEDRawDataCollection(), and read().

◆ L1EventID_

uint32_t FedRawDataInputSource::L1EventID_ = 0
private

Definition at line 124 of file FedRawDataInputSource.h.

Referenced by checkNext(), and read().

◆ loopModeIterationInc_

unsigned int FedRawDataInputSource::loopModeIterationInc_ = 0
private

Definition at line 109 of file FedRawDataInputSource.h.

Referenced by getFile().

◆ maxBufferedFiles_

unsigned int FedRawDataInputSource::maxBufferedFiles_
private

Definition at line 92 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

◆ MAXTCDSuTCAFEDID_

uint16_t FedRawDataInputSource::MAXTCDSuTCAFEDID_ = FEDNumbering::MAXTCDSuTCAFEDID
private

Definition at line 130 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and fillFEDRawDataCollection().

◆ MINTCDSuTCAFEDID_

uint16_t FedRawDataInputSource::MINTCDSuTCAFEDID_ = FEDNumbering::MINTCDSuTCAFEDID
private

Definition at line 129 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and fillFEDRawDataCollection().

◆ monlock_

std::mutex FedRawDataInputSource::monlock_
private

Definition at line 184 of file FedRawDataInputSource.h.

Referenced by getEventReport(), and reportEventsThisLumiInSource().

◆ mReader_

std::mutex FedRawDataInputSource::mReader_
private

Definition at line 154 of file FedRawDataInputSource.h.

Referenced by readSupervisor(), readWorker(), and ~FedRawDataInputSource().

◆ mWakeup_

std::mutex FedRawDataInputSource::mWakeup_
private

Definition at line 173 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

◆ nStreams_

unsigned int FedRawDataInputSource::nStreams_ = 0
private

Definition at line 169 of file FedRawDataInputSource.h.

Referenced by read().

◆ numBuffers_

unsigned int FedRawDataInputSource::numBuffers_
private

Definition at line 91 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

◆ numConcurrentReads_

unsigned int FedRawDataInputSource::numConcurrentReads_
private

Definition at line 93 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readWorker(), and ~FedRawDataInputSource().

◆ processHistoryID_

edm::ProcessHistoryID FedRawDataInputSource::processHistoryID_
private

Definition at line 119 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), maybeOpenNewLumiSection(), and read().

◆ quit_threads_

std::atomic<bool> FedRawDataInputSource::quit_threads_
private

Definition at line 158 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readSupervisor(), threadError(), and ~FedRawDataInputSource().

◆ readBlocks_

unsigned int FedRawDataInputSource::readBlocks_
private

Definition at line 90 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readNextChunkIntoBuffer(), and readWorker().

◆ readingFilesCount_

std::atomic<unsigned int> FedRawDataInputSource::readingFilesCount_
private

Definition at line 94 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), getNextEvent(), and readSupervisor().

◆ readSupervisorThread_

std::unique_ptr<std::thread> FedRawDataInputSource::readSupervisorThread_
private

Definition at line 145 of file FedRawDataInputSource.h.

Referenced by checkNext(), and ~FedRawDataInputSource().

◆ runNumber_

edm::RunNumber_t FedRawDataInputSource::runNumber_
private

Definition at line 111 of file FedRawDataInputSource.h.

Referenced by checkNext(), and FedRawDataInputSource().

◆ setExceptionState_

bool FedRawDataInputSource::setExceptionState_ = false
private

Definition at line 160 of file FedRawDataInputSource.h.

Referenced by exceptionState(), getNextEvent(), readSupervisor(), and readWorker().

◆ singleBufferMode_

bool FedRawDataInputSource::singleBufferMode_
private

Definition at line 177 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), getNextEvent(), and readSupervisor().

◆ sourceEventsReport_

std::map<unsigned int, unsigned int> FedRawDataInputSource::sourceEventsReport_
private

Definition at line 183 of file FedRawDataInputSource.h.

Referenced by getEventReport(), and reportEventsThisLumiInSource().

◆ startedSupervisorThread_

bool FedRawDataInputSource::startedSupervisorThread_ = false
private

Definition at line 144 of file FedRawDataInputSource.h.

Referenced by checkNext(), and ~FedRawDataInputSource().

◆ startupCv_

std::condition_variable FedRawDataInputSource::startupCv_
private

Definition at line 162 of file FedRawDataInputSource.h.

Referenced by checkNext(), FedRawDataInputSource(), readSupervisor(), and readWorker().

◆ startupLock_

std::mutex FedRawDataInputSource::startupLock_
private

Definition at line 161 of file FedRawDataInputSource.h.

Referenced by checkNext(), FedRawDataInputSource(), readSupervisor(), and readWorker().

◆ streamFileTracker_

std::vector<int> FedRawDataInputSource::streamFileTracker_
private

Definition at line 168 of file FedRawDataInputSource.h.

Referenced by read().

◆ tcds_pointer_

unsigned char* FedRawDataInputSource::tcds_pointer_
private

Definition at line 125 of file FedRawDataInputSource.h.

Referenced by fillFEDRawDataCollection(), and read().

◆ testTCDSFEDRange_

const std::vector<unsigned int> FedRawDataInputSource::testTCDSFEDRange_
private

Definition at line 101 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

◆ thread_quit_signal

std::vector<unsigned int> FedRawDataInputSource::thread_quit_signal
private

Definition at line 159 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readSupervisor(), readWorker(), and ~FedRawDataInputSource().

◆ threadInit_

std::atomic<bool> FedRawDataInputSource::threadInit_
private

Definition at line 181 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and readWorker().

◆ tid_active_

std::vector<unsigned int> FedRawDataInputSource::tid_active_
private

Definition at line 156 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readSupervisor(), and readWorker().

◆ useFileBroker_

bool FedRawDataInputSource::useFileBroker_
private

Definition at line 103 of file FedRawDataInputSource.h.

Referenced by checkNext(), FedRawDataInputSource(), maybeOpenNewLumiSection(), and readSupervisor().

◆ useL1EventID_

const bool FedRawDataInputSource::useL1EventID_
private

Definition at line 100 of file FedRawDataInputSource.h.

Referenced by read().

◆ verifyChecksum_

const bool FedRawDataInputSource::verifyChecksum_
private

Definition at line 99 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

◆ workerJob_

std::vector<ReaderInfo> FedRawDataInputSource::workerJob_
private

Definition at line 149 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readSupervisor(), and readWorker().

◆ workerPool_

tbb::concurrent_queue<unsigned int> FedRawDataInputSource::workerPool_
private

Definition at line 148 of file FedRawDataInputSource.h.

Referenced by readSupervisor(), and readWorker().

◆ workerThreads_

std::vector<std::thread*> FedRawDataInputSource::workerThreads_
private

Definition at line 146 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), readSupervisor(), and ~FedRawDataInputSource().

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
mps_fire.i
i
Definition: mps_fire.py:428
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
evf::FastMonState::inSupNewFileWaitChunkCopying
Definition: FastMonitoringService.h:115
evf::EvFDaqDirector::getEoLSFilePathOnBU
std::string getEoLSFilePathOnBU(const unsigned int ls) const
Definition: EvFDaqDirector.cc:476
FEDNumbering::MINTriggerEGTPFEDID
Definition: FEDNumbering.h:63
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
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
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
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
FedRawDataInputSource::defPath_
std::string defPath_
Definition: FedRawDataInputSource.h:86
newFWLiteAna.found
found
Definition: newFWLiteAna.py:118
edm::LogWarning
Log< level::Warning, false > LogWarning
Definition: MessageLogger.h:122
edm::RawInputSource
Definition: RawInputSource.h:17
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
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
Utilities.operator
operator
Definition: Utilities.py:24
FedRawDataInputSource::readBlocks_
unsigned int readBlocks_
Definition: FedRawDataInputSource.h:90
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::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
FedRawDataInputSource::filesToDelete_
std::list< std::pair< int, std::unique_ptr< InputFile > > > filesToDelete_
Definition: FedRawDataInputSource.h:165
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
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
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
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
first
auto first
Definition: CAHitNtupletGeneratorKernelsImpl.h:125
edm::EventAuxiliary
Definition: EventAuxiliary.h:14
FedRawDataInputSource::startedSupervisorThread_
bool startedSupervisorThread_
Definition: FedRawDataInputSource.h:144
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
evf::FastMonState::inSupNewFile
Definition: FastMonitoringService.h:112
InputChunk::fileIndex_
unsigned int fileIndex_
Definition: FedRawDataInputSource.h:194
LogDebug
#define LogDebug(id)
Definition: MessageLogger.h:233
edm::InputSource::setRunAuxiliary
void setRunAuxiliary(RunAuxiliary *rp)
Definition: InputSource.h:333
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
FedRawDataInputSource::alwaysStartFromFirstLS_
const bool alwaysStartFromFirstLS_
Definition: FedRawDataInputSource.h:98
edm::InputSource::productRegistryUpdate
ProductRegistry & productRegistryUpdate()
Definition: InputSource.h:330
FedRawDataInputSource::eventChunkBlock_
unsigned int eventChunkBlock_
Definition: FedRawDataInputSource.h:89
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
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::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
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
evf::FastMonState::inSupWaitFreeChunk
Definition: FastMonitoringService.h:104
FedRawDataInputSource::mReader_
std::mutex mReader_
Definition: FedRawDataInputSource.h:154
FedRawDataInputSource::eventsThisLumi_
unsigned int eventsThisLumi_
Definition: FedRawDataInputSource.h:126
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
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
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
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
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
edm::TypeID
Definition: TypeID.h:22
eostools.move
def move(src, dest)
Definition: eostools.py:511
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
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
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
FedRawDataInputSource::fileDescriptor_
int fileDescriptor_
Definition: FedRawDataInputSource.h:178
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
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
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
FEDHeader
Definition: FEDHeader.h:14
FedRawDataInputSource::MAXTCDSuTCAFEDID_
uint16_t MAXTCDSuTCAFEDID_
Definition: FedRawDataInputSource.h:130
lumi
Definition: LumiSectionData.h:20
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
FedRawDataInputSource::read
void read(edm::EventPrincipal &eventPrincipal) override
Definition: FedRawDataInputSource.cc:613
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
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
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
edm::Timestamp
Definition: Timestamp.h:30
Generated for CMSSW Reference Manual by doxygen 1.8.16