#include <FedRawDataInputSource.h>

Inheritance diagram for FedRawDataInputSource:

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 const >	branchIDListHelper () const
	Accessors for branchIDListHelper. More...

std::shared_ptr< BranchIDListHelper > &	branchIDListHelper ()

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 (ParameterSet const &, InputSourceDescription const &)
	Constructor. More...

	InputSource (InputSource const &)=delete

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 const >	processBlockHelper () const
	Accessors for processBlockHelper. More...

std::shared_ptr< ProcessBlockHelper > &	processBlockHelper ()

ProcessConfiguration const &	processConfiguration () const
	Accessor for Process Configuration. More...

std::string const &	processGUID () const
	Accessor for global process identifier. More...

ProcessHistoryRegistry const &	processHistoryRegistry () const
	Accessors for process history registry. More...

ProcessHistoryRegistry &	processHistoryRegistry ()

ProcessingMode	processingMode () const
	RunsLumisAndEvents (default), RunsAndLumis, or Runs. More...

std::shared_ptr< ProductRegistry const >	productRegistry () const
	Accessors for product registry. More...

std::shared_ptr< ProductRegistry > &	productRegistry ()

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...

void	readEvent (EventPrincipal &ep, StreamContext &)
	Read next event. More...

bool	readEvent (EventPrincipal &ep, EventID const &, StreamContext &)
	Read a specific 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 const >	thinnedAssociationsHelper () const
	Accessors for thinnedAssociationsHelper. More...

std::shared_ptr< ThinnedAssociationsHelper > &	thinnedAssociationsHelper ()

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 51 of file FedRawDataInputSource.cc.

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_.

     : edm::RawInputSource(pset, desc),
       defPath_(pset.getUntrackedParameter<std::string>("buDefPath", "")),
       eventChunkSize_(pset.getUntrackedParameter<unsigned int>("eventChunkSize", 32) * 1048576),
       eventChunkBlock_(pset.getUntrackedParameter<unsigned int>("eventChunkBlock", 32) * 1048576),
       numBuffers_(pset.getUntrackedParameter<unsigned int>("numBuffers", 2)),
       maxBufferedFiles_(pset.getUntrackedParameter<unsigned int>("maxBufferedFiles", 2)),
       getLSFromFilename_(pset.getUntrackedParameter<bool>("getLSFromFilename", true)),
       alwaysStartFromFirstLS_(pset.getUntrackedParameter<bool>("alwaysStartFromFirstLS", false)),
       verifyChecksum_(pset.getUntrackedParameter<bool>("verifyChecksum", true)),
       useL1EventID_(pset.getUntrackedParameter<bool>("useL1EventID", false)),
       testTCDSFEDRange_(
           pset.getUntrackedParameter<std::vector<unsigned int>>("testTCDSFEDRange", std::vector<unsigned int>())),
       fileNames_(pset.getUntrackedParameter<std::vector<std::string>>("fileNames", std::vector<std::string>())),
       fileListMode_(pset.getUntrackedParameter<bool>("fileListMode", false)),
       fileListLoopMode_(pset.getUntrackedParameter<bool>("fileListLoopMode", false)),
       runNumber_(edm::Service<evf::EvFDaqDirector>()->getRunNumber()),
       daqProvenanceHelper_(edm::TypeID(typeid(FEDRawDataCollection))),
       eventID_(),
       processHistoryID_(),
       currentLumiSection_(0),
       tcds_pointer_(nullptr),
       eventsThisLumi_(0) {
   char thishost[256];
   gethostname(thishost, 255);
   edm::LogInfo("FedRawDataInputSource") << "Construction. read-ahead chunk size -: " << std::endl
                                         << (eventChunkSize_ / 1048576) << " MB on host " << thishost;
 
   if (!testTCDSFEDRange_.empty()) {
     if (testTCDSFEDRange_.size() != 2) {
       throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection")
           << "Invalid TCDS Test FED range parameter";
     }
     MINTCDSuTCAFEDID_ = testTCDSFEDRange_[0];
     MAXTCDSuTCAFEDID_ = testTCDSFEDRange_[1];
   }
 
   long autoRunNumber = -1;
   if (fileListMode_) {
     autoRunNumber = initFileList();
     if (!fileListLoopMode_) {
       if (autoRunNumber < 0)
         throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource") << "Run number not found from filename";
       //override run number
       runNumber_ = (edm::RunNumber_t)autoRunNumber;
       edm::Service<evf::EvFDaqDirector>()->overrideRunNumber((unsigned int)autoRunNumber);
     }
   }
 
   processHistoryID_ = daqProvenanceHelper_.daqInit(productRegistryUpdate(), processHistoryRegistryForUpdate());
   setNewRun();
   //todo:autodetect from file name (assert if names differ)
   setRunAuxiliary(new edm::RunAuxiliary(runNumber_, edm::Timestamp::beginOfTime(), edm::Timestamp::invalidTimestamp()));
 
   //make sure that chunk size is N * block size
   assert(eventChunkSize_ >= eventChunkBlock_);
   readBlocks_ = eventChunkSize_ / eventChunkBlock_;
   if (readBlocks_ * eventChunkBlock_ != eventChunkSize_)
     eventChunkSize_ = readBlocks_ * eventChunkBlock_;
 
   if (!numBuffers_)
     throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource")
         << "no reading enabled with numBuffers parameter 0";
 
   numConcurrentReads_ = numBuffers_ - 1;
   singleBufferMode_ = !(numBuffers_ > 1);
   readingFilesCount_ = 0;
 
   if (!crc32c_hw_test())
     edm::LogError("FedRawDataInputSource::FedRawDataInputSource") << "Intel crc32c checksum computation unavailable";
 
   //get handles to DaqDirector and FastMonitoringService because getting them isn't possible in readSupervisor thread
   if (fileListMode_) {
     try {
       fms_ = static_cast<evf::FastMonitoringService*>(edm::Service<evf::MicroStateService>().operator->());
     } catch (cms::Exception const&) {
       edm::LogInfo("FedRawDataInputSource") << "No FastMonitoringService found in the configuration";
     }
   } else {
     fms_ = static_cast<evf::FastMonitoringService*>(edm::Service<evf::MicroStateService>().operator->());
     if (!fms_) {
       throw cms::Exception("FedRawDataInputSource") << "FastMonitoringService not found";
     }
   }
 
   daqDirector_ = edm::Service<evf::EvFDaqDirector>().operator->();
   if (!daqDirector_)
     cms::Exception("FedRawDataInputSource") << "EvFDaqDirector not found";
 
   useFileBroker_ = daqDirector_->useFileBroker();
   if (useFileBroker_)
     edm::LogInfo("FedRawDataInputSource") << "EvFDaqDirector/Source configured to use file service";
   //set DaqDirector to delete files in preGlobalEndLumi callback
   daqDirector_->setDeleteTracking(&fileDeleteLock_, &filesToDelete_);
   if (fms_) {
     daqDirector_->setFMS(fms_);
     fms_->setInputSource(this);
     fms_->setInState(inInit);
     fms_->setInStateSup(inInit);
   }
   //should delete chunks when run stops
   for (unsigned int i = 0; i < numBuffers_; i++) {
     freeChunks_.push(new InputChunk(i, eventChunkSize_));
   }
 
   quit_threads_ = false;
 
   for (unsigned int i = 0; i < numConcurrentReads_; i++) {
     std::unique_lock<std::mutex> lk(startupLock_);
     //issue a memory fence here and in threads (constructor was segfaulting without this)
     thread_quit_signal.push_back(false);
     workerJob_.push_back(ReaderInfo(nullptr, nullptr));
     cvReader_.push_back(new std::condition_variable);
     tid_active_.push_back(0);
     threadInit_.store(false, std::memory_order_release);
     workerThreads_.push_back(new std::thread(&FedRawDataInputSource::readWorker, this, i));
     startupCv_.wait(lk);
   }
 
   runAuxiliary()->setProcessHistoryID(processHistoryID_);
 }

◆ ~FedRawDataInputSource()

FedRawDataInputSource::~FedRawDataInputSource ( )

override

Definition at line 173 of file FedRawDataInputSource.cc.

References currentFile_, cvReader_, evf::FastMonitoringService::exceptionDetected(), filesToDelete_, fms_, mps_fire::i, evf::FastMonitoringService::isExceptionOnData(), mReader_, numConcurrentReads_, quit_threads_, readSupervisorThread_, startedSupervisorThread_, thread_quit_signal, and workerThreads_.

                                               {
   quit_threads_ = true;
 
   //delete any remaining open files
   if (!fms_ || !fms_->exceptionDetected()) {
     for (auto it = filesToDelete_.begin(); it != filesToDelete_.end(); it++)
       it->second.reset();
   } else {
     //skip deleting files with exception
     for (auto it = filesToDelete_.begin(); it != filesToDelete_.end(); it++) {
       //it->second->unsetDeleteFile();
       if (fms_->isExceptionOnData(it->second->lumi_))
         it->second->unsetDeleteFile();
       else
         it->second.reset();
     }
     //disable deleting current file with exception
     if (currentFile_.get())
       if (fms_->isExceptionOnData(currentFile_->lumi_))
         currentFile_->unsetDeleteFile();
   }
 
   if (startedSupervisorThread_) {
     readSupervisorThread_->join();
   } else {
     //join aux threads in case the supervisor thread was not started
     for (unsigned int i = 0; i < workerThreads_.size(); i++) {
       std::unique_lock<std::mutex> lk(mReader_);
       thread_quit_signal[i] = true;
       cvReader_[i]->notify_one();
       lk.unlock();
       workerThreads_[i]->join();
       delete workerThreads_[i];
     }
   }
   for (unsigned int i = 0; i < numConcurrentReads_; i++)
     delete cvReader_[i];
   /*
   for (unsigned int i=0;i<numConcurrentReads_+1;i++) {
     InputChunk *ch;
     while (!freeChunks_.try_pop(ch)) {}
     delete ch;
   }
   */
 }

Member Function Documentation

◆ checkNext()

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

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 244 of file FedRawDataInputSource.cc.

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_.

                                                        {
   if (!startedSupervisorThread_) {
     //this thread opens new files and dispatches reading to worker readers
     //threadInit_.store(false,std::memory_order_release);
     std::unique_lock<std::mutex> lk(startupLock_);
     readSupervisorThread_ = std::make_unique<std::thread>(&FedRawDataInputSource::readSupervisor, this);
     startedSupervisorThread_ = true;
     startupCv_.wait(lk);
   }
   //signal hltd to start event accounting
   if (!currentLumiSection_)
     daqDirector_->createProcessingNotificationMaybe();
   setMonState(inWaitInput);
   switch (nextEvent()) {
     case evf::EvFDaqDirector::runEnded: {
       //maybe create EoL file in working directory before ending run
       struct stat buf;
       if (!useFileBroker_ && currentLumiSection_ > 0) {
         bool eolFound = (stat(daqDirector_->getEoLSFilePathOnBU(currentLumiSection_).c_str(), &buf) == 0);
         if (eolFound) {
           const std::string fuEoLS = daqDirector_->getEoLSFilePathOnFU(currentLumiSection_);
           bool found = (stat(fuEoLS.c_str(), &buf) == 0);
           if (!found) {
             daqDirector_->lockFULocal2();
             int eol_fd =
                 open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
             close(eol_fd);
             daqDirector_->unlockFULocal2();
           }
         }
       }
       //also create EoR file in FU data directory
       bool eorFound = (stat(daqDirector_->getEoRFilePathOnFU().c_str(), &buf) == 0);
       if (!eorFound) {
         int eor_fd = open(daqDirector_->getEoRFilePathOnFU().c_str(),
                           O_RDWR | O_CREAT,
                           S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
         close(eor_fd);
       }
       reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
       eventsThisLumi_ = 0;
       resetLuminosityBlockAuxiliary();
       edm::LogInfo("FedRawDataInputSource") << "----------------RUN ENDED----------------";
       return Next::kStop;
     }
     case evf::EvFDaqDirector::noFile: {
       //this is not reachable
       return Next::kEvent;
     }
     case evf::EvFDaqDirector::newLumi: {
       //std::cout << "--------------NEW LUMI---------------" << std::endl;
       return Next::kEvent;
     }
     default: {
       if (!getLSFromFilename_) {
         //get new lumi from file header
         if (event_->lumi() > currentLumiSection_) {
           reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
           eventsThisLumi_ = 0;
           maybeOpenNewLumiSection(event_->lumi());
         }
       }
       if (fileListMode_ || fileListLoopMode_)
         eventRunNumber_ = runNumber_;
       else
         eventRunNumber_ = event_->run();
       L1EventID_ = event_->event();
 
       setEventCached();
 
       return Next::kEvent;
     }
   }
 }

◆ exceptionState()

bool FedRawDataInputSource::exceptionState ( )

inlineprivate

Definition at line 68 of file FedRawDataInputSource.h.

References setExceptionState_.

Referenced by InputFile::advance().

68 { return setExceptionState_; }

FedRawDataInputSource::setExceptionState_

bool setExceptionState_

Definition: FedRawDataInputSource.h:160

◆ fillDescriptions()

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

static

Definition at line 219 of file FedRawDataInputSource.cc.

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

                                                                                        {
   edm::ParameterSetDescription desc;
   desc.setComment("File-based Filter Farm input source for reading raw data from BU ramdisk");
   desc.addUntracked<unsigned int>("eventChunkSize", 32)->setComment("Input buffer (chunk) size");
   desc.addUntracked<unsigned int>("eventChunkBlock", 32)
       ->setComment("Block size used in a single file read call (must be smaller or equal to buffer size)");
   desc.addUntracked<unsigned int>("numBuffers", 2)->setComment("Number of buffers used for reading input");
   desc.addUntracked<unsigned int>("maxBufferedFiles", 2)
       ->setComment("Maximum number of simultaneously buffered raw files");
   desc.addUntracked<unsigned int>("alwaysStartFromfirstLS", false)
       ->setComment("Force source to start from LS 1 if server provides higher lumisection number");
   desc.addUntracked<bool>("verifyChecksum", true)
       ->setComment("Verify event CRC-32C checksum of FRDv5 and higher or Adler32 with v3 and v4");
   desc.addUntracked<bool>("useL1EventID", false)
       ->setComment("Use L1 event ID from FED header if true or from TCDS FED if false");
   desc.addUntracked<std::vector<unsigned int>>("testTCDSFEDRange", std::vector<unsigned int>())
       ->setComment("[min, max] range to search for TCDS FED ID in test setup");
   desc.addUntracked<bool>("fileListMode", false)
       ->setComment("Use fileNames parameter to directly specify raw files to open");
   desc.addUntracked<std::vector<std::string>>("fileNames", std::vector<std::string>())
       ->setComment("file list used when fileListMode is enabled");
   desc.setAllowAnything();
   descriptions.add("source", desc);
 }

◆ fillFEDRawDataCollection()

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

private

Definition at line 702 of file FedRawDataInputSource.cc.

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().

                                                                                                              {
   edm::TimeValue_t time;
   timeval stv;
   gettimeofday(&stv, nullptr);
   time = stv.tv_sec;
   time = (time << 32) + stv.tv_usec;
   edm::Timestamp tstamp(time);
 
   uint32_t eventSize = event_->eventSize();
   unsigned char* event = (unsigned char*)event_->payload();
   GTPEventID_ = 0;
   tcds_pointer_ = nullptr;
   tcdsInRange = false;
   uint16_t selectedTCDSFed = 0;
   while (eventSize > 0) {
     assert(eventSize >= FEDTrailer::length);
     eventSize -= FEDTrailer::length;
     const FEDTrailer fedTrailer(event + eventSize);
     const uint32_t fedSize = fedTrailer.fragmentLength() << 3;  //trailer length counts in 8 bytes
     assert(eventSize >= fedSize - FEDHeader::length);
     eventSize -= (fedSize - FEDHeader::length);
     const FEDHeader fedHeader(event + eventSize);
     const uint16_t fedId = fedHeader.sourceID();
     if (fedId > FEDNumbering::MAXFEDID) {
       throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection") << "Out of range FED ID : " << fedId;
     } else if (fedId >= MINTCDSuTCAFEDID_ && fedId <= MAXTCDSuTCAFEDID_) {
       if (!selectedTCDSFed) {
         selectedTCDSFed = fedId;
         tcds_pointer_ = event + eventSize;
         if (fedId >= FEDNumbering::MINTCDSuTCAFEDID && fedId <= FEDNumbering::MAXTCDSuTCAFEDID) {
           tcdsInRange = true;
         }
       } else
         throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection")
             << "Second TCDS FED ID " << fedId << " found. First ID: " << selectedTCDSFed;
     }
     if (fedId == FEDNumbering::MINTriggerGTPFEDID) {
       if (evf::evtn::evm_board_sense(event + eventSize, fedSize))
         GTPEventID_ = evf::evtn::get(event + eventSize, true);
       else
         GTPEventID_ = evf::evtn::get(event + eventSize, false);
       //evf::evtn::evm_board_setformat(fedSize);
       const uint64_t gpsl = evf::evtn::getgpslow(event + eventSize);
       const uint64_t gpsh = evf::evtn::getgpshigh(event + eventSize);
       tstamp = edm::Timestamp(static_cast<edm::TimeValue_t>((gpsh << 32) + gpsl));
     }
     //take event ID from GTPE FED
     if (fedId == FEDNumbering::MINTriggerEGTPFEDID && GTPEventID_ == 0) {
       if (evf::evtn::gtpe_board_sense(event + eventSize)) {
         GTPEventID_ = evf::evtn::gtpe_get(event + eventSize);
       }
     }
     FEDRawData& fedData = rawData.FEDData(fedId);
     fedData.resize(fedSize);
     memcpy(fedData.data(), event + eventSize, fedSize);
   }
   assert(eventSize == 0);
 
   return tstamp;
 }

◆ getEventReport()

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

Definition at line 1556 of file FedRawDataInputSource.cc.

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

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

                                                                                                {
   std::lock_guard<std::mutex> lock(monlock_);
   auto itr = sourceEventsReport_.find(lumi);
   if (itr != sourceEventsReport_.end()) {
     std::pair<bool, unsigned int> ret(true, itr->second);
     if (erase)
       sourceEventsReport_.erase(itr);
     return ret;
   } else
     return std::pair<bool, unsigned int>(false, 0);
 }

◆ getFile()

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

private

Definition at line 1603 of file FedRawDataInputSource.cc.

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().

                                                                                        {
   if (fileListIndex_ < fileNames_.size()) {
     nextFile = fileNames_[fileListIndex_];
     if (nextFile.find("file://") == 0)
       nextFile = nextFile.substr(7);
     else if (nextFile.find("file:") == 0)
       nextFile = nextFile.substr(5);
     std::filesystem::path fileName = nextFile;
     std::string fileStem = fileName.stem().string();
     if (fileStem.find("ls"))
       fileStem = fileStem.substr(fileStem.find("ls") + 2);
     if (fileStem.find('_'))
       fileStem = fileStem.substr(0, fileStem.find('_'));
 
     if (!fileListLoopMode_)
       ls = std::stoul(fileStem);
     else  //always starting from LS 1 in loop mode
       ls = 1 + loopModeIterationInc_;
 
     //fsize = 0;
     //lockWaitTime = 0;
     fileListIndex_++;
     return evf::EvFDaqDirector::newFile;
   } else {
     if (!fileListLoopMode_)
       return evf::EvFDaqDirector::runEnded;
     else {
       //loop through files until interrupted
       loopModeIterationInc_++;
       fileListIndex_ = 0;
       return getFile(ls, nextFile, fsize, lockWaitTime);
     }
   }
 }

◆ getNextEvent()

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

inlineprivate

Definition at line 365 of file FedRawDataInputSource.cc.

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().

                                                                        {
   if (setExceptionState_)
     threadError();
   if (!currentFile_.get()) {
     evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
     setMonState(inWaitInput);
     if (!fileQueue_.try_pop(currentFile_)) {
       //sleep until wakeup (only in single-buffer mode) or timeout
       std::unique_lock<std::mutex> lkw(mWakeup_);
       if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout || !currentFile_.get())
         return evf::EvFDaqDirector::noFile;
     }
     status = currentFile_->status_;
     if (status == evf::EvFDaqDirector::runEnded) {
       setMonState(inRunEnd);
       currentFile_.reset();
       return status;
     } else if (status == evf::EvFDaqDirector::runAbort) {
       throw cms::Exception("FedRawDataInputSource::getNextEvent")
           << "Run has been aborted by the input source reader thread";
     } else if (status == evf::EvFDaqDirector::newLumi) {
       setMonState(inNewLumi);
       if (getLSFromFilename_) {
         if (currentFile_->lumi_ > currentLumiSection_) {
           reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
           eventsThisLumi_ = 0;
           maybeOpenNewLumiSection(currentFile_->lumi_);
         }
       } else {  //let this be picked up from next event
         status = evf::EvFDaqDirector::noFile;
       }
       currentFile_.reset();
       return status;
     } else if (status == evf::EvFDaqDirector::newFile) {
       currentFileIndex_++;
     } else
       assert(false);
   }
   setMonState(inProcessingFile);
 
   //file is empty
   if (!currentFile_->fileSize_) {
     readingFilesCount_--;
     //try to open new lumi
     assert(currentFile_->nChunks_ == 0);
     if (getLSFromFilename_)
       if (currentFile_->lumi_ > currentLumiSection_) {
         reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
         eventsThisLumi_ = 0;
         maybeOpenNewLumiSection(currentFile_->lumi_);
       }
     //immediately delete empty file
     currentFile_.reset();
     return evf::EvFDaqDirector::noFile;
   }
 
   //file is finished
   if (currentFile_->bufferPosition_ == currentFile_->fileSize_) {
     readingFilesCount_--;
     //release last chunk (it is never released elsewhere)
     freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_]);
     if (currentFile_->nEvents_ >= 0 && currentFile_->nEvents_ != int(currentFile_->nProcessed_)) {
       throw cms::Exception("FedRawDataInputSource::getNextEvent")
           << "Fully processed " << currentFile_->nProcessed_ << " from the file " << currentFile_->fileName_
           << " but according to BU JSON there should be " << currentFile_->nEvents_ << " events";
     }
     //try to wake up supervisor thread which might be sleeping waiting for the free chunk
     if (singleBufferMode_) {
       std::unique_lock<std::mutex> lkw(mWakeup_);
       cvWakeup_.notify_one();
     }
     bufferInputRead_ = 0;
     if (!daqDirector_->isSingleStreamThread() && !fileListMode_) {
       //put the file in pending delete list;
       std::unique_lock<std::mutex> lkw(fileDeleteLock_);
       filesToDelete_.push_back(std::pair<int, std::unique_ptr<InputFile>>(currentFileIndex_, std::move(currentFile_)));
     } else {
       //in single-thread and stream jobs, events are already processed
       currentFile_.reset();
     }
     return evf::EvFDaqDirector::noFile;
   }
 
   //handle RAW file header
   if (currentFile_->bufferPosition_ == 0 && currentFile_->rawHeaderSize_ > 0) {
     if (currentFile_->fileSize_ <= currentFile_->rawHeaderSize_) {
       if (currentFile_->fileSize_ < currentFile_->rawHeaderSize_)
         throw cms::Exception("FedRawDataInputSource::getNextEvent")
             << "Premature end of input file while reading file header";
 
       edm::LogWarning("FedRawDataInputSource")
           << "File with only raw header and no events received in LS " << currentFile_->lumi_;
       if (getLSFromFilename_)
         if (currentFile_->lumi_ > currentLumiSection_) {
           reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
           eventsThisLumi_ = 0;
           maybeOpenNewLumiSection(currentFile_->lumi_);
         }
     }
 
     //advance buffer position to skip file header (chunk will be acquired later)
     currentFile_->chunkPosition_ += currentFile_->rawHeaderSize_;
     currentFile_->bufferPosition_ += currentFile_->rawHeaderSize_;
   }
 
   //file is too short
   if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < FRDHeaderVersionSize[detectedFRDversion_]) {
     throw cms::Exception("FedRawDataInputSource::getNextEvent")
         << "Premature end of input file while reading event header";
   }
   if (singleBufferMode_) {
     //should already be there
     setMonState(inWaitChunk);
     while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
       usleep(10000);
       if (currentFile_->parent_->exceptionState() || setExceptionState_)
         currentFile_->parent_->threadError();
     }
     setMonState(inChunkReceived);
 
     unsigned char* dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
 
     //conditions when read amount is not sufficient for the header to fit
     if (!bufferInputRead_ || bufferInputRead_ < FRDHeaderVersionSize[detectedFRDversion_] ||
         eventChunkSize_ - currentFile_->chunkPosition_ < FRDHeaderVersionSize[detectedFRDversion_]) {
       readNextChunkIntoBuffer(currentFile_.get());
 
       if (detectedFRDversion_ == 0) {
         detectedFRDversion_ = *((uint16_t*)dataPosition);
         if (detectedFRDversion_ > FRDHeaderMaxVersion)
           throw cms::Exception("FedRawDataInputSource::getNextEvent")
               << "Unknown FRD version -: " << detectedFRDversion_;
         assert(detectedFRDversion_ >= 1);
       }
 
       //recalculate chunk position
       dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
       if (bufferInputRead_ < FRDHeaderVersionSize[detectedFRDversion_]) {
         throw cms::Exception("FedRawDataInputSource::getNextEvent")
             << "Premature end of input file while reading event header";
       }
     }
 
     event_ = std::make_unique<FRDEventMsgView>(dataPosition);
     if (event_->size() > eventChunkSize_) {
       throw cms::Exception("FedRawDataInputSource::getNextEvent")
           << " event id:" << event_->event() << " lumi:" << event_->lumi() << " run:" << event_->run()
           << " of size:" << event_->size() << " bytes does not fit into a chunk of size:" << eventChunkSize_
           << " bytes";
     }
 
     const uint32_t msgSize = event_->size() - FRDHeaderVersionSize[detectedFRDversion_];
 
     if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize) {
       throw cms::Exception("FedRawDataInputSource::getNextEvent")
           << "Premature end of input file while reading event data";
     }
     if (eventChunkSize_ - currentFile_->chunkPosition_ < msgSize) {
       readNextChunkIntoBuffer(currentFile_.get());
       //recalculate chunk position
       dataPosition = currentFile_->chunks_[0]->buf_ + currentFile_->chunkPosition_;
       event_ = std::make_unique<FRDEventMsgView>(dataPosition);
     }
     currentFile_->bufferPosition_ += event_->size();
     currentFile_->chunkPosition_ += event_->size();
     //last chunk is released when this function is invoked next time
 
   }
   //multibuffer mode:
   else {
     //wait for the current chunk to become added to the vector
     setMonState(inWaitChunk);
     while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
       usleep(10000);
       if (setExceptionState_)
         threadError();
     }
     setMonState(inChunkReceived);
 
     //check if header is at the boundary of two chunks
     chunkIsFree_ = false;
     unsigned char* dataPosition;
 
     //read header, copy it to a single chunk if necessary
     bool chunkEnd = currentFile_->advance(dataPosition, FRDHeaderVersionSize[detectedFRDversion_]);
 
     event_ = std::make_unique<FRDEventMsgView>(dataPosition);
     if (event_->size() > eventChunkSize_) {
       throw cms::Exception("FedRawDataInputSource::getNextEvent")
           << " event id:" << event_->event() << " lumi:" << event_->lumi() << " run:" << event_->run()
           << " of size:" << event_->size() << " bytes does not fit into a chunk of size:" << eventChunkSize_
           << " bytes";
     }
 
     const uint32_t msgSize = event_->size() - FRDHeaderVersionSize[detectedFRDversion_];
 
     if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize) {
       throw cms::Exception("FedRawDataInputSource::getNextEvent")
           << "Premature end of input file while reading event data";
     }
 
     if (chunkEnd) {
       //header was at the chunk boundary, we will have to move payload as well
       currentFile_->moveToPreviousChunk(msgSize, FRDHeaderVersionSize[detectedFRDversion_]);
       chunkIsFree_ = true;
     } else {
       //header was contiguous, but check if payload fits the chunk
       if (eventChunkSize_ - currentFile_->chunkPosition_ < msgSize) {
         //rewind to header start position
         currentFile_->rewindChunk(FRDHeaderVersionSize[detectedFRDversion_]);
         //copy event to a chunk start and move pointers
 
         setMonState(inWaitChunk);
 
         chunkEnd = currentFile_->advance(dataPosition, FRDHeaderVersionSize[detectedFRDversion_] + msgSize);
 
         setMonState(inChunkReceived);
 
         assert(chunkEnd);
         chunkIsFree_ = true;
         //header is moved
         event_ = std::make_unique<FRDEventMsgView>(dataPosition);
       } else {
         //everything is in a single chunk, only move pointers forward
         chunkEnd = currentFile_->advance(dataPosition, msgSize);
         assert(!chunkEnd);
         chunkIsFree_ = false;
       }
     }
   }  //end multibuffer mode
   setMonState(inChecksumEvent);
 
   if (verifyChecksum_ && event_->version() >= 5) {
     uint32_t crc = 0;
     crc = crc32c(crc, (const unsigned char*)event_->payload(), event_->eventSize());
     if (crc != event_->crc32c()) {
       if (fms_)
         fms_->setExceptionDetected(currentLumiSection_);
       throw cms::Exception("FedRawDataInputSource::getNextEvent")
           << "Found a wrong crc32c checksum: expected 0x" << std::hex << event_->crc32c() << " but calculated 0x"
           << crc;
     }
   } else if (verifyChecksum_ && event_->version() >= 3) {
     uint32_t adler = adler32(0L, Z_NULL, 0);
     adler = adler32(adler, (Bytef*)event_->payload(), event_->eventSize());
 
     if (adler != event_->adler32()) {
       if (fms_)
         fms_->setExceptionDetected(currentLumiSection_);
       throw cms::Exception("FedRawDataInputSource::getNextEvent")
           << "Found a wrong Adler32 checksum: expected 0x" << std::hex << event_->adler32() << " but calculated 0x"
           << adler;
     }
   }
   setMonState(inCachedEvent);
 
   currentFile_->nProcessed_++;
 
   return evf::EvFDaqDirector::sameFile;
 }

◆ initFileList()

long FedRawDataInputSource::initFileList ( )

private

Definition at line 1568 of file FedRawDataInputSource.cc.

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

Referenced by FedRawDataInputSource().

                                          {
   std::sort(fileNames_.begin(), fileNames_.end(), [](std::string a, std::string b) {
     if (a.rfind('/') != std::string::npos)
       a = a.substr(a.rfind('/'));
     if (b.rfind('/') != std::string::npos)
       b = b.substr(b.rfind('/'));
     return b > a;
   });
 
   if (!fileNames_.empty()) {
     //get run number from first file in the vector
     std::filesystem::path fileName = fileNames_[0];
     std::string fileStem = fileName.stem().string();
     if (fileStem.find("file://") == 0)
       fileStem = fileStem.substr(7);
     else if (fileStem.find("file:") == 0)
       fileStem = fileStem.substr(5);
     auto end = fileStem.find('_');
 
     if (fileStem.find("run") == 0) {
       std::string runStr = fileStem.substr(3, end - 3);
       try {
         //get long to support test run numbers < 2^32
         long rval = std::stol(runStr);
         edm::LogInfo("FedRawDataInputSource") << "Autodetected run number in fileListMode -: " << rval;
         return rval;
       } catch (const std::exception&) {
         edm::LogWarning("FedRawDataInputSource")
             << "Unable to autodetect run number in fileListMode from file -: " << fileName;
       }
     }
   }
   return -1;
 }

◆ maybeOpenNewLumiSection()

void FedRawDataInputSource::maybeOpenNewLumiSection ( const uint32_t lumiSection )

private

Definition at line 319 of file FedRawDataInputSource.cc.

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().

                                                                               {
   if (!luminosityBlockAuxiliary() || luminosityBlockAuxiliary()->luminosityBlock() != lumiSection) {
     if (!useFileBroker_) {
       if (currentLumiSection_ > 0) {
         const std::string fuEoLS = daqDirector_->getEoLSFilePathOnFU(currentLumiSection_);
         struct stat buf;
         bool found = (stat(fuEoLS.c_str(), &buf) == 0);
         if (!found) {
           daqDirector_->lockFULocal2();
           int eol_fd =
               open(fuEoLS.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
           close(eol_fd);
           daqDirector_->createBoLSFile(lumiSection, false);
           daqDirector_->unlockFULocal2();
         }
       } else
         daqDirector_->createBoLSFile(lumiSection, true);  //needed for initial lumisection
     }
 
     currentLumiSection_ = lumiSection;
 
     resetLuminosityBlockAuxiliary();
 
     timeval tv;
     gettimeofday(&tv, nullptr);
     const edm::Timestamp lsopentime((unsigned long long)tv.tv_sec * 1000000 + (unsigned long long)tv.tv_usec);
 
     edm::LuminosityBlockAuxiliary* lumiBlockAuxiliary = new edm::LuminosityBlockAuxiliary(
         runAuxiliary()->run(), lumiSection, lsopentime, edm::Timestamp::invalidTimestamp());
 
     setLuminosityBlockAuxiliary(lumiBlockAuxiliary);
     luminosityBlockAuxiliary()->setProcessHistoryID(processHistoryID_);
 
     edm::LogInfo("FedRawDataInputSource") << "New lumi section was opened. LUMI -: " << lumiSection;
   }
 }

◆ nextEvent()

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

inlineprivate

Definition at line 356 of file FedRawDataInputSource.cc.

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

Referenced by checkNext().

                                                                     {
   evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
   while ((status = getNextEvent()) == evf::EvFDaqDirector::noFile) {
     if (edm::shutdown_flag.load(std::memory_order_relaxed))
       break;
   }
   return status;
 }

◆ read()

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

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 626 of file FedRawDataInputSource.cc.

References printConversionInfo::aux, edm::DaqProvenanceHelper::branchDescription(), checkEvery_, chunkIsFree_, runTheMatrix::const, currentFile_, currentFileIndex_, currentLumiSection_, daqProvenanceHelper_, edm::DaqProvenanceHelper::dummyProvenance(), event_, eventID_, eventRunNumber_, eventsThisLumi_, Exception, fileDeleteLock_, fileListLoopMode_, filesToDelete_, fillFEDRawDataCollection(), fms_, freeChunks_, GTPEventID_, mps_fire::i, evf::FastMonState::inNoRequest, evf::FastMonState::inReadCleanup, evf::FastMonState::inReadEvent, evf::FastMonitoringService::isExceptionOnData(), 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().

                                                                   {
   setMonState(inReadEvent);
   std::unique_ptr<FEDRawDataCollection> rawData(new FEDRawDataCollection);
   bool tcdsInRange;
   edm::Timestamp tstamp = fillFEDRawDataCollection(*rawData, tcdsInRange);
 
   if (useL1EventID_) {
     eventID_ = edm::EventID(eventRunNumber_, currentLumiSection_, L1EventID_);
     edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, event_->isRealData(), edm::EventAuxiliary::PhysicsTrigger);
     aux.setProcessHistoryID(processHistoryID_);
     makeEvent(eventPrincipal, aux);
   } else if (tcds_pointer_ == nullptr) {
     if (!GTPEventID_) {
       throw cms::Exception("FedRawDataInputSource::read")
           << "No TCDS or GTP FED in event with FEDHeader EID -: " << L1EventID_;
     }
     eventID_ = edm::EventID(eventRunNumber_, currentLumiSection_, GTPEventID_);
     edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, event_->isRealData(), edm::EventAuxiliary::PhysicsTrigger);
     aux.setProcessHistoryID(processHistoryID_);
     makeEvent(eventPrincipal, aux);
   } else {
     const FEDHeader fedHeader(tcds_pointer_);
     tcds::Raw_v1 const* tcds = reinterpret_cast<tcds::Raw_v1 const*>(tcds_pointer_ + FEDHeader::length);
     edm::EventAuxiliary aux =
         evf::evtn::makeEventAuxiliary(tcds,
                                       eventRunNumber_,
                                       currentLumiSection_,
                                       event_->isRealData(),
                                       static_cast<edm::EventAuxiliary::ExperimentType>(fedHeader.triggerType()),
                                       processGUID(),
                                       !fileListLoopMode_,
                                       !tcdsInRange);
     aux.setProcessHistoryID(processHistoryID_);
     makeEvent(eventPrincipal, aux);
   }
 
   std::unique_ptr<edm::WrapperBase> edp(new edm::Wrapper<FEDRawDataCollection>(std::move(rawData)));
 
   eventPrincipal.put(daqProvenanceHelper_.branchDescription(), std::move(edp), daqProvenanceHelper_.dummyProvenance());
 
   eventsThisLumi_++;
   setMonState(inReadCleanup);
 
   //resize vector if needed
   while (streamFileTracker_.size() <= eventPrincipal.streamID())
     streamFileTracker_.push_back(-1);
 
   streamFileTracker_[eventPrincipal.streamID()] = currentFileIndex_;
 
   //this old file check runs no more often than every 10 events
   if (!((currentFile_->nProcessed_ - 1) % (checkEvery_))) {
     //delete files that are not in processing
     std::unique_lock<std::mutex> lkw(fileDeleteLock_);
     auto it = filesToDelete_.begin();
     while (it != filesToDelete_.end()) {
       bool fileIsBeingProcessed = false;
       for (unsigned int i = 0; i < nStreams_; i++) {
         if (it->first == streamFileTracker_.at(i)) {
           fileIsBeingProcessed = true;
           break;
         }
       }
       if (!fileIsBeingProcessed && (!fms_ || !fms_->isExceptionOnData(it->second->lumi_))) {
         std::string fileToDelete = it->second->fileName_;
         it = filesToDelete_.erase(it);
       } else
         it++;
     }
   }
   if (chunkIsFree_)
     freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_ - 1]);
   chunkIsFree_ = false;
   setMonState(inNoRequest);
   return;
 }

◆ readNextChunkIntoBuffer()

void FedRawDataInputSource::readNextChunkIntoBuffer ( InputFile * file )

private

Definition at line 1476 of file FedRawDataInputSource.cc.

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

Referenced by getNextEvent().

                                                                    {
   uint32_t existingSize = 0;
 
   if (fileDescriptor_ < 0) {
     bufferInputRead_ = 0;
     if (file->rawFd_ == -1) {
       fileDescriptor_ = open(file->fileName_.c_str(), O_RDONLY);
       if (file->rawHeaderSize_)
         lseek(fileDescriptor_, file->rawHeaderSize_, SEEK_SET);
     } else
       fileDescriptor_ = file->rawFd_;
 
     //skip header size in destination buffer (chunk position was already adjusted)
     bufferInputRead_ += file->rawHeaderSize_;
     existingSize += file->rawHeaderSize_;
 
     if (fileDescriptor_ >= 0)
       LogDebug("FedRawDataInputSource") << "opened file -: " << std::endl << file->fileName_;
     else {
       throw cms::Exception("FedRawDataInputSource:readNextChunkIntoBuffer")
           << "failed to open file " << std::endl
           << file->fileName_ << " fd:" << fileDescriptor_;
     }
     //fill chunk (skipping file header if present)
     for (unsigned int i = 0; i < readBlocks_; i++) {
       const ssize_t last = ::read(fileDescriptor_,
                                   (void*)(file->chunks_[0]->buf_ + existingSize),
                                   eventChunkBlock_ - (i == readBlocks_ - 1 ? existingSize : 0));
       bufferInputRead_ += last;
       existingSize += last;
     }
 
   } else {
     //continue reading
     if (file->chunkPosition_ == 0) {  //in the rare case the last byte barely fit
       for (unsigned int i = 0; i < readBlocks_; i++) {
         const ssize_t last = ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSize), eventChunkBlock_);
         bufferInputRead_ += last;
         existingSize += last;
       }
     } else {
       //event didn't fit in last chunk, so leftover must be moved to the beginning and completed
       uint32_t existingSizeLeft = eventChunkSize_ - file->chunkPosition_;
       memmove((void*)file->chunks_[0]->buf_, file->chunks_[0]->buf_ + file->chunkPosition_, existingSizeLeft);
 
       //calculate amount of data that can be added
       const uint32_t blockcount = file->chunkPosition_ / eventChunkBlock_;
       const uint32_t leftsize = file->chunkPosition_ % eventChunkBlock_;
 
       for (uint32_t i = 0; i < blockcount; i++) {
         const ssize_t last =
             ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSizeLeft), eventChunkBlock_);
         bufferInputRead_ += last;
         existingSizeLeft += last;
       }
       if (leftsize) {
         const ssize_t last = ::read(fileDescriptor_, (void*)(file->chunks_[0]->buf_ + existingSizeLeft), leftsize);
         bufferInputRead_ += last;
       }
       file->chunkPosition_ = 0;  //data was moved to beginning of the chunk
     }
   }
   if (bufferInputRead_ == file->fileSize_) {  // no more data in this file
     if (fileDescriptor_ != -1) {
       LogDebug("FedRawDataInputSource") << "Closing input file -: " << std::endl << file->fileName_;
       close(fileDescriptor_);
       file->rawFd_ = fileDescriptor_ = -1;
     }
   }
 }

◆ readSupervisor()

void FedRawDataInputSource::readSupervisor ( )

private

Definition at line 765 of file FedRawDataInputSource.cc.

Referenced by checkNext().

                                            {
   bool stop = false;
   unsigned int currentLumiSection = 0;
   //threadInit_.exchange(true,std::memory_order_acquire);
 
   {
     std::unique_lock<std::mutex> lk(startupLock_);
     startupCv_.notify_one();
   }
 
   uint32_t ls = 0;
   uint32_t monLS = 1;
   uint32_t lockCount = 0;
   uint64_t sumLockWaitTimeUs = 0.;
 
   while (!stop) {
     //wait for at least one free thread and chunk
     int counter = 0;
 
     while ((workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty() ||
            readingFilesCount_ >= maxBufferedFiles_) {
       //report state to monitoring
       if (fms_) {
         bool copy_active = false;
         for (auto j : tid_active_)
           if (j)
             copy_active = true;
         if (readingFilesCount_ >= maxBufferedFiles_)
           setMonStateSup(inSupFileLimit);
         else if (freeChunks_.empty()) {
           if (copy_active)
             setMonStateSup(inSupWaitFreeChunkCopying);
           else
             setMonStateSup(inSupWaitFreeChunk);
         } else {
           if (copy_active)
             setMonStateSup(inSupWaitFreeThreadCopying);
           else
             setMonStateSup(inSupWaitFreeThread);
         }
       }
       std::unique_lock<std::mutex> lkw(mWakeup_);
       //sleep until woken up by condition or a timeout
       if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout) {
         counter++;
         //if (!(counter%50)) edm::LogInfo("FedRawDataInputSource") << "No free chunks or threads...";
         LogDebug("FedRawDataInputSource") << "No free chunks or threads...";
       } else {
         assert(!(workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty());
       }
       if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
         stop = true;
         break;
       }
     }
     //if this is reached, there are enough buffers and threads to proceed or processing is instructed to stop
 
     if (stop)
       break;
 
     //look for a new file
     std::string nextFile;
     uint32_t fileSizeIndex;
     int64_t fileSizeFromMetadata;
 
     if (fms_) {
       setMonStateSup(inSupBusy);
       fms_->startedLookingForFile();
     }
 
     evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
     uint16_t rawHeaderSize = 0;
     uint32_t lsFromRaw = 0;
     int32_t serverEventsInNewFile = -1;
     int rawFd = -1;
 
     int backoff_exp = 0;
 
     //entering loop which tries to grab new file from ramdisk
     while (status == evf::EvFDaqDirector::noFile) {
       //check if hltd has signalled to throttle input
       counter = 0;
       while (daqDirector_->inputThrottled()) {
         if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed))
           break;
         setMonStateSup(inThrottled);
         if (!(counter % 50))
           edm::LogWarning("FedRawDataInputSource") << "Input throttled detected, reading files is paused...";
         usleep(100000);
         counter++;
       }
 
       if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
         stop = true;
         break;
       }
 
       assert(rawFd == -1);
       uint64_t thisLockWaitTimeUs = 0.;
       setMonStateSup(inSupLockPolling);
       if (fileListMode_) {
         //return LS if LS not set, otherwise return file
         status = getFile(ls, nextFile, fileSizeIndex, thisLockWaitTimeUs);
         if (status == evf::EvFDaqDirector::newFile) {
           if (evf::EvFDaqDirector::parseFRDFileHeader(nextFile,
                                                       rawFd,
                                                       rawHeaderSize,
                                                       lsFromRaw,
                                                       serverEventsInNewFile,
                                                       fileSizeFromMetadata,
                                                       false,
                                                       false,
                                                       false) != 0) {
             //error
             setExceptionState_ = true;
             stop = true;
             break;
           }
           if (!getLSFromFilename_)
             ls = lsFromRaw;
         }
       } else if (!useFileBroker_)
         status = daqDirector_->updateFuLock(
             ls, nextFile, fileSizeIndex, rawHeaderSize, thisLockWaitTimeUs, setExceptionState_);
       else {
         status = daqDirector_->getNextFromFileBroker(currentLumiSection,
                                                      ls,
                                                      nextFile,
                                                      rawFd,
                                                      rawHeaderSize,
                                                      serverEventsInNewFile,
                                                      fileSizeFromMetadata,
                                                      thisLockWaitTimeUs);
       }
 
       setMonStateSup(inSupBusy);
 
       //cycle through all remaining LS even if no files get assigned
       if (currentLumiSection != ls && status == evf::EvFDaqDirector::runEnded)
         status = evf::EvFDaqDirector::noFile;
 
       //monitoring of lock wait time
       if (thisLockWaitTimeUs > 0.)
         sumLockWaitTimeUs += thisLockWaitTimeUs;
       lockCount++;
       if (ls > monLS) {
         monLS = ls;
         if (lockCount)
           if (fms_)
             fms_->reportLockWait(monLS, sumLockWaitTimeUs, lockCount);
         lockCount = 0;
         sumLockWaitTimeUs = 0;
       }
 
       //check again for any remaining index/EoLS files after EoR file is seen
       if (status == evf::EvFDaqDirector::runEnded && !fileListMode_ && !useFileBroker_) {
         setMonStateSup(inRunEnd);
         usleep(100000);
         //now all files should have appeared in ramdisk, check again if any raw files were left behind
         status = daqDirector_->updateFuLock(
             ls, nextFile, fileSizeIndex, rawHeaderSize, thisLockWaitTimeUs, setExceptionState_);
         if (currentLumiSection != ls && status == evf::EvFDaqDirector::runEnded)
           status = evf::EvFDaqDirector::noFile;
       }
 
       if (status == evf::EvFDaqDirector::runEnded) {
         std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runEnded));
         fileQueue_.push(std::move(inf));
         stop = true;
         break;
       }
 
       //error from filelocking function
       if (status == evf::EvFDaqDirector::runAbort) {
         std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runAbort, 0));
         fileQueue_.push(std::move(inf));
         stop = true;
         break;
       }
       //queue new lumisection
       if (getLSFromFilename_) {
         if (ls > currentLumiSection) {
           if (!useFileBroker_) {
             //file locking
             //setMonStateSup(inSupNewLumi);
             currentLumiSection = ls;
             std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, currentLumiSection));
             fileQueue_.push(std::move(inf));
           } else {
             //new file service
             if (currentLumiSection == 0 && !alwaysStartFromFirstLS_) {
               if (daqDirector_->getStartLumisectionFromEnv() > 1) {
                 //start transitions from LS specified by env, continue if not reached
                 if (ls < daqDirector_->getStartLumisectionFromEnv()) {
                   //skip file if from earlier LS than specified by env
                   if (rawFd != -1) {
                     close(rawFd);
                     rawFd = -1;
                   }
                   status = evf::EvFDaqDirector::noFile;
                   continue;
                 } else {
                   std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, ls));
                   fileQueue_.push(std::move(inf));
                 }
               } else if (ls < 100) {
                 //look at last LS file on disk to start from that lumisection (only within first 100 LS)
                 unsigned int lsToStart = daqDirector_->getLumisectionToStart();
 
                 for (unsigned int nextLS = std::min(lsToStart, ls); nextLS <= ls; nextLS++) {
                   std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, nextLS));
                   fileQueue_.push(std::move(inf));
                 }
               } else {
                 //start from current LS
                 std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, ls));
                 fileQueue_.push(std::move(inf));
               }
             } else {
               //queue all lumisections after last one seen to avoid gaps
               for (unsigned int nextLS = currentLumiSection + 1; nextLS <= ls; nextLS++) {
                 std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::newLumi, nextLS));
                 fileQueue_.push(std::move(inf));
               }
             }
             currentLumiSection = ls;
           }
         }
         //else
         if (currentLumiSection > 0 && ls < currentLumiSection) {
           edm::LogError("FedRawDataInputSource")
               << "Got old LS (" << ls << ") file from EvFDAQDirector! Expected LS:" << currentLumiSection
               << ". Aborting execution." << std::endl;
           if (rawFd != -1)
             close(rawFd);
           rawFd = -1;
           std::unique_ptr<InputFile> inf(new InputFile(evf::EvFDaqDirector::runAbort, 0));
           fileQueue_.push(std::move(inf));
           stop = true;
           break;
         }
       }
 
       int dbgcount = 0;
       if (status == evf::EvFDaqDirector::noFile) {
         setMonStateSup(inSupNoFile);
         dbgcount++;
         if (!(dbgcount % 20))
           LogDebug("FedRawDataInputSource") << "No file for me... sleep and try again...";
         if (!useFileBroker_)
           usleep(100000);
         else {
           backoff_exp = std::min(4, backoff_exp);  // max 1.6 seconds
           //backoff_exp=0; // disabled!
           int sleeptime = (int)(100000. * pow(2, backoff_exp));
           usleep(sleeptime);
           backoff_exp++;
         }
       } else
         backoff_exp = 0;
     }
     //end of file grab loop, parse result
     if (status == evf::EvFDaqDirector::newFile) {
       setMonStateSup(inSupNewFile);
       LogDebug("FedRawDataInputSource") << "The director says to grab -: " << nextFile;
 
       std::string rawFile;
       //file service will report raw extension
       if (useFileBroker_ || rawHeaderSize)
         rawFile = nextFile;
       else {
         std::filesystem::path rawFilePath(nextFile);
         rawFile = rawFilePath.replace_extension(".raw").string();
       }
 
       struct stat st;
       int stat_res = stat(rawFile.c_str(), &st);
       if (stat_res == -1) {
         edm::LogError("FedRawDataInputSource") << "Can not stat file (" << errno << "):-" << rawFile << std::endl;
         setExceptionState_ = true;
         break;
       }
       uint64_t fileSize = st.st_size;
 
       if (fms_) {
         setMonStateSup(inSupBusy);
         fms_->stoppedLookingForFile(ls);
         setMonStateSup(inSupNewFile);
       }
       int eventsInNewFile;
       if (fileListMode_) {
         if (fileSize == 0)
           eventsInNewFile = 0;
         else
           eventsInNewFile = -1;
       } else {
         std::string empty;
         if (!useFileBroker_) {
           if (rawHeaderSize) {
             int rawFdEmpty = -1;
             uint16_t rawHeaderCheck;
             bool fileFound;
             eventsInNewFile = daqDirector_->grabNextJsonFromRaw(
                 nextFile, rawFdEmpty, rawHeaderCheck, fileSizeFromMetadata, fileFound, 0, true);
             assert(fileFound && rawHeaderCheck == rawHeaderSize);
             daqDirector_->unlockFULocal();
           } else
             eventsInNewFile = daqDirector_->grabNextJsonFileAndUnlock(nextFile);
         } else
           eventsInNewFile = serverEventsInNewFile;
         assert(eventsInNewFile >= 0);
         assert((eventsInNewFile > 0) ==
                (fileSize > rawHeaderSize));  //file without events must be empty or contain only header
       }
 
       if (!singleBufferMode_) {
         //calculate number of needed chunks
         unsigned int neededChunks = fileSize / eventChunkSize_;
         if (fileSize % eventChunkSize_)
           neededChunks++;
 
         std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
                                                               ls,
                                                               rawFile,
                                                               !fileListMode_,
                                                               rawFd,
                                                               fileSize,
                                                               rawHeaderSize,
                                                               neededChunks,
                                                               eventsInNewFile,
                                                               this));
         readingFilesCount_++;
         auto newInputFilePtr = newInputFile.get();
         fileQueue_.push(std::move(newInputFile));
 
         for (unsigned int i = 0; i < neededChunks; i++) {
           if (fms_) {
             bool copy_active = false;
             for (auto j : tid_active_)
               if (j)
                 copy_active = true;
             if (copy_active)
               setMonStateSup(inSupNewFileWaitThreadCopying);
             else
               setMonStateSup(inSupNewFileWaitThread);
           }
           //get thread
           unsigned int newTid = 0xffffffff;
           while (!workerPool_.try_pop(newTid)) {
             usleep(100000);
             if (quit_threads_.load(std::memory_order_relaxed)) {
               stop = true;
               break;
             }
           }
 
           if (fms_) {
             bool copy_active = false;
             for (auto j : tid_active_)
               if (j)
                 copy_active = true;
             if (copy_active)
               setMonStateSup(inSupNewFileWaitChunkCopying);
             else
               setMonStateSup(inSupNewFileWaitChunk);
           }
           InputChunk* newChunk = nullptr;
           while (!freeChunks_.try_pop(newChunk)) {
             usleep(100000);
             if (quit_threads_.load(std::memory_order_relaxed)) {
               stop = true;
               break;
             }
           }
 
           if (newChunk == nullptr) {
             //return unused tid if we received shutdown (nullptr chunk)
             if (newTid != 0xffffffff)
               workerPool_.push(newTid);
             stop = true;
             break;
           }
           if (stop)
             break;
           setMonStateSup(inSupNewFile);
 
           std::unique_lock<std::mutex> lk(mReader_);
 
           unsigned int toRead = eventChunkSize_;
           if (i == neededChunks - 1 && fileSize % eventChunkSize_)
             toRead = fileSize % eventChunkSize_;
           newChunk->reset(i * eventChunkSize_, toRead, i);
 
           workerJob_[newTid].first = newInputFilePtr;
           workerJob_[newTid].second = newChunk;
 
           //wake up the worker thread
           cvReader_[newTid]->notify_one();
         }
       } else {
         if (!eventsInNewFile) {
           if (rawFd) {
             close(rawFd);
             rawFd = -1;
           }
           //still queue file for lumi update
           std::unique_lock<std::mutex> lkw(mWakeup_);
           //TODO: also file with only file header fits in this edge case. Check if read correctly in single buffer mode
           std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
                                                                 ls,
                                                                 rawFile,
                                                                 !fileListMode_,
                                                                 rawFd,
                                                                 fileSize,
                                                                 rawHeaderSize,
                                                                 (rawHeaderSize > 0),
                                                                 0,
                                                                 this));
           readingFilesCount_++;
           fileQueue_.push(std::move(newInputFile));
           cvWakeup_.notify_one();
           break;
         }
         //in single-buffer mode put single chunk in the file and let the main thread read the file
         InputChunk* newChunk;
         //should be available immediately
         while (!freeChunks_.try_pop(newChunk)) {
           usleep(100000);
           if (quit_threads_.load(std::memory_order_relaxed))
             break;
         }
 
         std::unique_lock<std::mutex> lkw(mWakeup_);
 
         unsigned int toRead = eventChunkSize_;
         if (fileSize % eventChunkSize_)
           toRead = fileSize % eventChunkSize_;
         newChunk->reset(0, toRead, 0);
         newChunk->readComplete_ = true;
 
         //push file and wakeup main thread
         std::unique_ptr<InputFile> newInputFile(new InputFile(evf::EvFDaqDirector::FileStatus::newFile,
                                                               ls,
                                                               rawFile,
                                                               !fileListMode_,
                                                               rawFd,
                                                               fileSize,
                                                               rawHeaderSize,
                                                               1,
                                                               eventsInNewFile,
                                                               this));
         newInputFile->chunks_[0] = newChunk;
         readingFilesCount_++;
         fileQueue_.push(std::move(newInputFile));
         cvWakeup_.notify_one();
       }
     }
   }
   setMonStateSup(inRunEnd);
   //make sure threads finish reading
   unsigned numFinishedThreads = 0;
   while (numFinishedThreads < workerThreads_.size()) {
     unsigned tid = 0;
     while (!workerPool_.try_pop(tid)) {
       usleep(10000);
     }
     std::unique_lock<std::mutex> lk(mReader_);
     thread_quit_signal[tid] = true;
     cvReader_[tid]->notify_one();
     numFinishedThreads++;
   }
   for (unsigned int i = 0; i < workerThreads_.size(); i++) {
     workerThreads_[i]->join();
     delete workerThreads_[i];
   }
 }

◆ readWorker()

void FedRawDataInputSource::readWorker ( unsigned int tid )

private

Definition at line 1242 of file FedRawDataInputSource.cc.

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, SiStripPI::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().

                                                        {
   bool init = true;
   threadInit_.exchange(true, std::memory_order_acquire);
 
   while (true) {
     tid_active_[tid] = false;
     std::unique_lock<std::mutex> lk(mReader_);
     workerJob_[tid].first = nullptr;
     workerJob_[tid].first = nullptr;
 
     assert(!thread_quit_signal[tid]);  //should never get it here
     workerPool_.push(tid);
 
     if (init) {
       std::unique_lock<std::mutex> lk(startupLock_);
       init = false;
       startupCv_.notify_one();
     }
     cvReader_[tid]->wait(lk);
 
     if (thread_quit_signal[tid])
       return;
     tid_active_[tid] = true;
 
     InputFile* file;
     InputChunk* chunk;
 
     assert(workerJob_[tid].first != nullptr && workerJob_[tid].second != nullptr);
 
     file = workerJob_[tid].first;
     chunk = workerJob_[tid].second;
 
     //skip reading initial header size in first chunk if inheriting file descriptor (already set at appropriate position)
     unsigned int bufferLeft = (chunk->offset_ == 0 && file->rawFd_ != -1) ? file->rawHeaderSize_ : 0;
 
     //if only one worker thread exists, use single fd for all operations
     //if more worker threads exist, use rawFd_ for only the first read operation and then close file
     int fileDescriptor;
     bool fileOpenedHere = false;
 
     if (numConcurrentReads_ == 1) {
       fileDescriptor = file->rawFd_;
       if (fileDescriptor == -1) {
         fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
         fileOpenedHere = true;
         file->rawFd_ = fileDescriptor;
       }
     } else {
       if (chunk->offset_ == 0) {
         fileDescriptor = file->rawFd_;
         file->rawFd_ = -1;
         if (fileDescriptor == -1) {
           fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
           fileOpenedHere = true;
         }
       } else {
         fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
         fileOpenedHere = true;
       }
     }
 
     if (fileDescriptor < 0) {
       edm::LogError("FedRawDataInputSource") << "readWorker failed to open file -: " << file->fileName_
                                              << " fd:" << fileDescriptor << " error: " << strerror(errno);
       setExceptionState_ = true;
       continue;
     }
     if (fileOpenedHere) {  //fast forward to this chunk position
       off_t pos = 0;
       pos = lseek(fileDescriptor, chunk->offset_, SEEK_SET);
       if (pos == -1) {
         edm::LogError("FedRawDataInputSource")
             << "readWorker failed to seek file -: " << file->fileName_ << " fd:" << fileDescriptor << " to offset "
             << chunk->offset_ << " error: " << strerror(errno);
         setExceptionState_ = true;
         continue;
       }
     }
 
     LogDebug("FedRawDataInputSource") << "Reader thread opened file -: TID: " << tid << " file: " << file->fileName_
                                       << " at offset " << lseek(fileDescriptor, 0, SEEK_CUR);
 
     unsigned int skipped = bufferLeft;
     auto start = std::chrono::high_resolution_clock::now();
     for (unsigned int i = 0; i < readBlocks_; i++) {
       ssize_t last;
 
       //protect against reading into next block
       last = ::read(
           fileDescriptor, (void*)(chunk->buf_ + bufferLeft), std::min(chunk->usedSize_ - bufferLeft, eventChunkBlock_));
 
       if (last < 0) {
         edm::LogError("FedRawDataInputSource") << "readWorker failed to read file -: " << file->fileName_
                                                << " fd:" << fileDescriptor << " error: " << strerror(errno);
         setExceptionState_ = true;
         break;
       }
       if (last > 0)
         bufferLeft += last;
       if (last < eventChunkBlock_) {  //last read
         //check if this is last block, then total read size must match file size
         if (!(chunk->usedSize_ - skipped == i * eventChunkBlock_ + last)) {
           edm::LogError("FedRawDataInputSource")
               << "readWorker failed to read file -: " << file->fileName_ << " fd:" << fileDescriptor << " last:" << last
               << " expectedChunkSize:" << chunk->usedSize_
               << " readChunkSize:" << (skipped + i * eventChunkBlock_ + last) << " skipped:" << skipped
               << " block:" << (i + 1) << "/" << readBlocks_ << " error: " << strerror(errno);
           setExceptionState_ = true;
         }
         break;
       }
     }
     if (setExceptionState_)
       continue;
 
     auto end = std::chrono::high_resolution_clock::now();
     auto diff = end - start;
     std::chrono::milliseconds msec = std::chrono::duration_cast<std::chrono::milliseconds>(diff);
     LogDebug("FedRawDataInputSource") << " finished reading block -: " << (bufferLeft >> 20) << " MB"
                                       << " in " << msec.count() << " ms (" << (bufferLeft >> 20) / double(msec.count())
                                       << " GB/s)";
 
     if (chunk->offset_ + bufferLeft == file->fileSize_) {  //file reading finished using same fd
       close(fileDescriptor);
       fileDescriptor = -1;
       if (numConcurrentReads_ == 1)
         file->rawFd_ = -1;
     }
     if (numConcurrentReads_ > 1 && fileDescriptor != -1)
       close(fileDescriptor);
 
     //detect FRD event version. Skip file Header if it exists
     if (detectedFRDversion_ == 0 && chunk->offset_ == 0) {
       detectedFRDversion_ = *((uint16_t*)(chunk->buf_ + file->rawHeaderSize_));
     }
     assert(detectedFRDversion_ <= FRDHeaderMaxVersion);
     chunk->readComplete_ =
         true;  //this is atomic to secure the sequential buffer fill before becoming available for processing)
     file->chunks_[chunk->fileIndex_] = chunk;  //put the completed chunk in the file chunk vector at predetermined index
   }
 }

◆ reportEventsThisLumiInSource()

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

private

Definition at line 1547 of file FedRawDataInputSource.cc.

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

Referenced by checkNext(), and getNextEvent().

                                                                                                {
   std::lock_guard<std::mutex> lock(monlock_);
   auto itr = sourceEventsReport_.find(lumi);
   if (itr != sourceEventsReport_.end())
     itr->second += events;
   else
     sourceEventsReport_[lumi] = events;
 }

◆ rewind_()

void FedRawDataInputSource::rewind_ ( )

overrideprivatevirtual

Reimplemented from edm::InputSource.

Definition at line 763 of file FedRawDataInputSource.cc.

763 {}

◆ setMonState()

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

inlineprotected

Definition at line 1389 of file FedRawDataInputSource.cc.

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

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

                                                                               {
   if (fms_)
     fms_->setInState(state);
 }

◆ setMonStateSup()

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

inlineprotected

Definition at line 1394 of file FedRawDataInputSource.cc.

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

Referenced by readSupervisor().

                                                                                  {
   if (fms_)
     fms_->setInStateSup(state);
 }

◆ threadError()

void FedRawDataInputSource::threadError ( )

private

Definition at line 1384 of file FedRawDataInputSource.cc.

References Exception, and quit_threads_.

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

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

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(), read(), and ~FedRawDataInputSource().

◆ 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(), read(), readSupervisor(), setMonState(), setMonStateSup(), and ~FedRawDataInputSource().

◆ 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().

Public Member Functions

Static Public Member Functions

Protected Member Functions

Private Types

Private Member Functions

Private Attributes

Friends

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ ReaderInfo

Constructor & Destructor Documentation

◆ FedRawDataInputSource()

◆ ~FedRawDataInputSource()

Member Function Documentation

◆ checkNext()

◆ exceptionState()

◆ fillDescriptions()

◆ fillFEDRawDataCollection()

◆ getEventReport()

◆ getFile()

◆ getNextEvent()

◆ initFileList()

◆ maybeOpenNewLumiSection()

◆ nextEvent()

◆ read()

◆ readNextChunkIntoBuffer()

◆ readSupervisor()

◆ readWorker()

◆ reportEventsThisLumiInSource()

◆ rewind_()

◆ setMonState()

◆ setMonStateSup()

◆ threadError()

Friends And Related Function Documentation

◆ InputChunk

◆ InputFile

Member Data Documentation

◆ alwaysStartFromFirstLS_

◆ bufferInputRead_

◆ checkEvery_

◆ chunkIsFree_

◆ currentFile_

◆ currentFileIndex_

◆ currentLumiSection_

◆ cvReader_

◆ cvWakeup_

◆ daqDirector_

◆ daqProvenanceHelper_

◆ defPath_

◆ detectedFRDversion_

◆ event_

◆ eventChunkBlock_

◆ eventChunkSize_

◆ eventID_

◆ eventRunNumber_

◆ eventsThisLumi_

◆ eventsThisRun_

◆ fileDeleteLock_

◆ fileDescriptor_

◆ fileListIndex_

◆ fileListLoopMode_

◆ fileListMode_

◆ fileNames_

◆ fileNamesToDelete_

◆ fileQueue_

◆ filesToDelete_

◆ fms_

◆ freeChunks_

◆ fuOutputDir_

◆ getLSFromFilename_

◆ GTPEventID_

◆ L1EventID_

◆ loopModeIterationInc_

◆ maxBufferedFiles_

◆ MAXTCDSuTCAFEDID_

◆ MINTCDSuTCAFEDID_

◆ monlock_

◆ mReader_

◆ mWakeup_