#include <DAQSource.h>

Inheritance diagram for DAQSource:

Public Member Functions
int	currentLumiSection () const

	DAQSource (edm::ParameterSet const &, edm::InputSourceDescription const &)

int	eventRunNumber () const

bool	fileListLoopMode ()

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

void	makeEventWrapper (edm::EventPrincipal &eventPrincipal, edm::EventAuxiliary &aux)

edm::ProcessHistoryID &	processHistoryID ()

bool	useL1EventID () const

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

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

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)

void	switchTo (std::shared_ptr< ProductRegistry > iOther)
	switch to a different ProductRegistry. More...

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)

ItemTypeInfo	state () const

Private Types
typedef std::pair< RawInputFile , InputChunk >	ReaderInfo

Private Member Functions
void	dataArranger ()

bool	exceptionState ()

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

evf::EvFDaqDirector::FileStatus	getNextDataBlock ()

evf::EvFDaqDirector::FileStatus	getNextEventFromDataBlock ()

long	initFileList ()

void	maybeOpenNewLumiSection (const uint32_t lumiSection)

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_

unsigned int	checkEvery_ = 10

bool	chunkIsFree_ = false

std::unique_ptr< RawInputFile >	currentFile_

int	currentFileIndex_ = -1

unsigned int	currentLumiSection_

std::vector< std::unique_ptr< std::condition_variable > >	cvReader_

std::condition_variable	cvWakeup_

evf::EvFDaqDirector *	daqDirector_ = nullptr

std::unique_ptr< std::thread >	dataArrangerThread_

std::shared_ptr< DataMode >	dataMode_

const std::string	dataModeConfig_

uint64_t	eventChunkBlock_

uint64_t	eventChunkSize_

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_

tbb::concurrent_queue< std::unique_ptr< RawInputFile > >	fileQueue_

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

evf::FastMonitoringService *	fms_ = nullptr

tbb::concurrent_queue< InputChunk * >	freeChunks_

std::string	fuOutputDir_

uint32_t	GTPEventID_ = 0

std::vector< std::string >	listFileNames_

unsigned int	loopModeIterationInc_ = 0

unsigned int	maxBufferedFiles_

uint64_t	maxChunkSize_

std::mutex	monlock_

std::mutex	mReader_

std::mutex	mWakeup_

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_

std::default_random_engine	rng_

edm::RunNumber_t	runNumber_

bool	setExceptionState_ = false

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

bool	startedSupervisorThread_ = false

std::condition_variable	startupCv_

std::mutex	startupLock_

std::vector< int >	streamFileTracker_

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

class	RawInputFile

Additional Inherited Members
Public Types inherited from edm::RawInputSource
enum	Next { Next::kEvent, Next::kFile, Next::kStop }

Public Types inherited from edm::InputSource
enum	ItemPosition : char { ItemPosition::Invalid, ItemPosition::LastItemToBeMerged, ItemPosition::NotLastItemToBeMerged }

enum	ItemType : char { ItemType::IsInvalid, ItemType::IsStop, ItemType::IsFile, ItemType::IsRun, ItemType::IsLumi, ItemType::IsEvent, ItemType::IsRepeat, ItemType::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 41 of file DAQSource.h.

Member Typedef Documentation

◆ ReaderInfo

typedef std::pair<RawInputFile*, InputChunk*> DAQSource::ReaderInfo

private

Definition at line 132 of file DAQSource.h.

Constructor & Destructor Documentation

◆ DAQSource()

DAQSource::DAQSource	(	edm::ParameterSet const &	pset,
		edm::InputSourceDescription const &	desc
	)

explicit

Definition at line 32 of file DAQSource.cc.

References cms::cuda::assert(), edm::Timestamp::beginOfTime(), crc32c_hw_test(), cvReader_, daqDirector_, dataMode_, dataModeConfig_, eventChunkBlock_, eventChunkSize_, Exception, fileDeleteLock_, fileListLoopMode_, fileListMode_, filesToDelete_, fms_, freeChunks_, evf::EvFDaqDirector::getBUBaseDirs(), evf::EvFDaqDirector::getBUBaseDirsNSources(), mps_fire::i, evf::FastMonState::inInit, initFileList(), InputChunk, edm::Timestamp::invalidTimestamp(), maxChunkSize_, FEDNumbering::MAXTCDSuTCAFEDID, FEDNumbering::MINTCDSuTCAFEDID, numBuffers_, numConcurrentReads_, Utilities::operator, evf::EvFDaqDirector::overrideRunNumber(), processHistoryID_, edm::InputSource::processHistoryRegistryForUpdate(), edm::InputSource::productRegistryUpdate(), muonDTDigis_cfi::pset, quit_threads_, readBlocks_, readingFilesCount_, readWorker(), edm::InputSource::runAuxiliary(), runNumber_, evf::EvFDaqDirector::runString(), evf::EvFDaqDirector::setDeleteTracking(), evf::EvFDaqDirector::setFMS(), evf::FastMonitoringService::setInputSource(), evf::FastMonitoringService::setInState(), evf::FastMonitoringService::setInStateSup(), edm::InputSource::setNewRun(), edm::InputSource::setRunAuxiliary(), startupCv_, startupLock_, testTCDSFEDRange_, thread_quit_signal, tid_active_, workerJob_, and workerThreads_.

     : edm::RawInputSource(pset, desc),
       dataModeConfig_(pset.getUntrackedParameter<std::string>("dataMode")),
       eventChunkSize_(uint64_t(pset.getUntrackedParameter<unsigned int>("eventChunkSize")) << 20),
       maxChunkSize_(uint64_t(pset.getUntrackedParameter<unsigned int>("maxChunkSize")) << 20),
       eventChunkBlock_(uint64_t(pset.getUntrackedParameter<unsigned int>("eventChunkBlock")) << 20),
       numBuffers_(pset.getUntrackedParameter<unsigned int>("numBuffers")),
       maxBufferedFiles_(pset.getUntrackedParameter<unsigned int>("maxBufferedFiles")),
       alwaysStartFromFirstLS_(pset.getUntrackedParameter<bool>("alwaysStartFromFirstLS", false)),
       verifyChecksum_(pset.getUntrackedParameter<bool>("verifyChecksum")),
       useL1EventID_(pset.getUntrackedParameter<bool>("useL1EventID")),
       testTCDSFEDRange_(pset.getUntrackedParameter<std::vector<unsigned int>>("testTCDSFEDRange")),
       listFileNames_(pset.getUntrackedParameter<std::vector<std::string>>("fileNames")),
       fileListMode_(pset.getUntrackedParameter<bool>("fileListMode")),
       fileListLoopMode_(pset.getUntrackedParameter<bool>("fileListLoopMode", false)),
       runNumber_(edm::Service<evf::EvFDaqDirector>()->getRunNumber()),
       processHistoryID_(),
       currentLumiSection_(0),
       eventsThisLumi_(0),
       rng_(std::chrono::system_clock::now().time_since_epoch().count()) {
   char thishost[256];
   gethostname(thishost, 255);
 
   if (maxChunkSize_ == 0)
     maxChunkSize_ = eventChunkSize_;
   else if (maxChunkSize_ < eventChunkSize_)
     throw cms::Exception("DAQSource::DAQSource") << "maxChunkSize must be equal or larger than eventChunkSize";
 
   if (eventChunkBlock_ == 0)
     eventChunkBlock_ = eventChunkSize_;
   else if (eventChunkBlock_ > eventChunkSize_)
     throw cms::Exception("DAQSource::DAQSource") << "eventChunkBlock must be equal or smaller than eventChunkSize";
 
   edm::LogInfo("DAQSource") << "Construction. read-ahead chunk size -: " << std::endl
                             << (eventChunkSize_ >> 20) << " MB on host " << thishost << " in mode " << dataModeConfig_;
 
   uint16_t MINTCDSuTCAFEDID = FEDNumbering::MINTCDSuTCAFEDID;
   uint16_t MAXTCDSuTCAFEDID = FEDNumbering::MAXTCDSuTCAFEDID;
 
   if (!testTCDSFEDRange_.empty()) {
     if (testTCDSFEDRange_.size() != 2) {
       throw cms::Exception("DAQSource::DAQSource") << "Invalid TCDS Test FED range parameter";
     }
     MINTCDSuTCAFEDID = testTCDSFEDRange_[0];
     MAXTCDSuTCAFEDID = testTCDSFEDRange_[1];
   }
 
   //load mode class based on parameter
   if (dataModeConfig_ == "FRD") {
     dataMode_.reset(new DataModeFRD(this));
   } else if (dataModeConfig_ == "FRDStriped") {
     dataMode_.reset(new DataModeFRDStriped(this));
   } else if (dataModeConfig_ == "ScoutingRun3") {
     dataMode_.reset(new DataModeScoutingRun3(this));
   } else
     throw cms::Exception("DAQSource::DAQSource") << "Unknown data mode " << dataModeConfig_;
 
   daqDirector_ = edm::Service<evf::EvFDaqDirector>().operator->();
 
   dataMode_->setTCDSSearchRange(MINTCDSuTCAFEDID, MAXTCDSuTCAFEDID);
   dataMode_->setTesting(pset.getUntrackedParameter<bool>("testing", false));
 
   long autoRunNumber = -1;
   if (fileListMode_) {
     autoRunNumber = initFileList();
     if (!fileListLoopMode_) {
       if (autoRunNumber < 0)
         throw cms::Exception("DAQSource::DAQSource") << "Run number not found from filename";
       //override run number
       runNumber_ = (edm::RunNumber_t)autoRunNumber;
       daqDirector_->overrideRunNumber((unsigned int)autoRunNumber);
     }
   }
 
   dataMode_->makeDirectoryEntries(
       daqDirector_->getBUBaseDirs(), daqDirector_->getBUBaseDirsNSources(), daqDirector_->runString());
 
   auto& daqProvenanceHelpers = dataMode_->makeDaqProvenanceHelpers();
   for (const auto& daqProvenanceHelper : daqProvenanceHelpers)
     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("DAQSource::DAQSource") << "no reading enabled with numBuffers parameter 0";
 
   numConcurrentReads_ = numBuffers_ - 1;
   assert(numBuffers_ > 1);
   readingFilesCount_ = 0;
 
   if (!crc32c_hw_test())
     edm::LogError("DAQSource::DAQSource") << "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::FastMonitoringService>().operator->());
     } catch (cms::Exception const&) {
       edm::LogInfo("DAQSource") << "No FastMonitoringService found in the configuration";
     }
   } else {
     fms_ = static_cast<evf::FastMonitoringService*>(edm::Service<evf::FastMonitoringService>().operator->());
     if (!fms_) {
       throw cms::Exception("DAQSource") << "FastMonitoringService not found";
     }
   }
 
   daqDirector_ = edm::Service<evf::EvFDaqDirector>().operator->();
   if (!daqDirector_)
     cms::Exception("DAQSource") << "EvFDaqDirector not found";
 
   edm::LogInfo("DAQSource") << "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(eventChunkSize_));
   }
 
   quit_threads_ = false;
 
   //prepare data shared by threads
   for (unsigned int i = 0; i < numConcurrentReads_; i++) {
     thread_quit_signal.push_back(false);
     workerJob_.push_back(ReaderInfo(nullptr, nullptr));
     cvReader_.push_back(std::make_unique<std::condition_variable>());
     tid_active_.push_back(0);
   }
 
   //start threads
   for (unsigned int i = 0; i < numConcurrentReads_; i++) {
     //wait for each thread to complete initialization
     std::unique_lock<std::mutex> lk(startupLock_);
     workerThreads_.push_back(new std::thread(&DAQSource::readWorker, this, i));
     startupCv_.wait(lk);
   }
 
   runAuxiliary()->setProcessHistoryID(processHistoryID_);
 }

◆ ~DAQSource()

DAQSource::~DAQSource ( )

override

Definition at line 185 of file DAQSource.cc.

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

                       {
   quit_threads_ = true;
 
   //delete any remaining open files
   if (!fms_ || !fms_->exceptionDetected()) {
     std::unique_lock<std::mutex> lkw(fileDeleteLock_);
     for (auto it = filesToDelete_.begin(); it != filesToDelete_.end(); it++)
       it->second.reset();
   } else {
     //skip deleting files with exception
     std::unique_lock<std::mutex> lkw(fileDeleteLock_);
     for (auto it = filesToDelete_.begin(); it != filesToDelete_.end(); it++) {
       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];
     }
   }
 }

Member Function Documentation

◆ checkNext()

edm::RawInputSource::Next DAQSource::checkNext ( )

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 254 of file DAQSource.cc.

References visDQMUpload::buf, evf::EvFDaqDirector::createProcessingNotificationMaybe(), currentLumiSection_, daqDirector_, dataMode_, eventRunNumber_, eventsThisLumi_, fileListLoopMode_, fileListMode_, evf::EvFDaqDirector::getEoRFilePathOnFU(), getNextDataBlock(), getNextEventFromDataBlock(), evf::FastMonState::inWaitInput, edm::RawInputSource::kEvent, edm::RawInputSource::kStop, svgfig::load(), evf::EvFDaqDirector::newLumi, evf::EvFDaqDirector::noFile, readSupervisor(), readSupervisorThread_, reportEventsThisLumiInSource(), edm::InputSource::resetLuminosityBlockAuxiliary(), evf::EvFDaqDirector::runEnded, runNumber_, edm::InputSource::setEventCached(), setMonState(), edm::shutdown_flag, startedSupervisorThread_, startupCv_, startupLock_, edm_modernize_messagelogger::stat, and mps_update::status.

                                            {
   if (!startedSupervisorThread_) {
     std::unique_lock<std::mutex> lk(startupLock_);
 
     //this thread opens new files and dispatches reading to worker readers
     readSupervisorThread_ = std::make_unique<std::thread>(&DAQSource::readSupervisor, this);
     startedSupervisorThread_ = true;
 
     startupCv_.wait(lk);
   }
 
   //signal hltd to start event accounting
   if (!currentLumiSection_)
     daqDirector_->createProcessingNotificationMaybe();
   setMonState(inWaitInput);
 
   auto nextEvent = [this]() {
     auto getNextEvent = [this]() {
       //for some models this is always true (if one event is one block)
       if (dataMode_->dataBlockCompleted()) {
         return getNextDataBlock();
       } else {
         return getNextEventFromDataBlock();
       }
     };
 
     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;
   };
 
   switch (nextEvent()) {
     case evf::EvFDaqDirector::runEnded: {
       //maybe create EoL file in working directory before ending run
       struct stat buf;
       //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("DAQSource") << "----------------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 (fileListMode_ || fileListLoopMode_)
         eventRunNumber_ = runNumber_;
       else
         eventRunNumber_ = dataMode_->run();
 
       setEventCached();
 
       return Next::kEvent;
     }
   }
 }

◆ currentLumiSection()

int DAQSource::currentLumiSection ( ) const

inline

Definition at line 52 of file DAQSource.h.

References currentLumiSection_.

Referenced by readSupervisor().

52 { return currentLumiSection_; }

DAQSource::currentLumiSection_

unsigned int currentLumiSection_

Definition: DAQSource.h:119

◆ dataArranger()

void DAQSource::dataArranger ( )

private

Definition at line 591 of file DAQSource.cc.

591 {}

◆ eventRunNumber()

int DAQSource::eventRunNumber ( ) const

inline

Definition at line 53 of file DAQSource.h.

References eventRunNumber_.

53 { return eventRunNumber_; }

DAQSource::eventRunNumber_

uint32_t eventRunNumber_

Definition: DAQSource.h:120

◆ exceptionState()

bool DAQSource::exceptionState ( )

inlineprivate

Definition at line 78 of file DAQSource.h.

References setExceptionState_.

Referenced by RawInputFile::advance().

78 { return setExceptionState_; }

DAQSource::setExceptionState_

bool setExceptionState_

Definition: DAQSource.h:154

◆ fileListLoopMode()

bool DAQSource::fileListLoopMode ( )

inline

Definition at line 57 of file DAQSource.h.

References fileListLoopMode_.

57 { return fileListLoopMode_; }

DAQSource::fileListLoopMode_

const bool fileListLoopMode_

Definition: DAQSource.h:111

◆ fillDescriptions()

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

static

Definition at line 223 of file DAQSource.cc.

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

                                                                            {
   edm::ParameterSetDescription desc;
   desc.setComment("File-based Filter Farm input source for reading raw data from BU ramdisk (unified)");
   desc.addUntracked<std::string>("dataMode", "FRD")->setComment("Data mode: event 'FRD', 'FRSStriped', 'ScoutingRun2'");
   desc.addUntracked<unsigned int>("eventChunkSize", 64)->setComment("Input buffer (chunk) size");
   desc.addUntracked<unsigned int>("maxChunkSize", 0)
       ->setComment("Maximum chunk size allowed if buffer is resized to fit data. If 0 is specifier, use chunk size");
   desc.addUntracked<unsigned int>("eventChunkBlock", 0)
       ->setComment(
           "Block size used in a single file read call (must be smaller or equal to the initial chunk buffer size). If "
           "0 is specified, use chunk 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);
 }

◆ getEventReport()

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

Definition at line 1390 of file DAQSource.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 DAQSource::getFile	(	unsigned int &	ls,
		std::string &	nextFile,
		uint64_t &	lockWaitTime
	)

private

Definition at line 1436 of file DAQSource.cc.

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

Referenced by readSupervisor().

                                                                                                             {
   if (fileListIndex_ < listFileNames_.size()) {
     nextFile = listFileNames_[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, lockWaitTime);
     }
   }
 }

◆ getNextDataBlock()

evf::EvFDaqDirector::FileStatus DAQSource::getNextDataBlock ( )

inlineprivate

Definition at line 373 of file DAQSource.cc.

References cms::cuda::assert(), chunkIsFree_, currentFile_, currentFileIndex_, currentLumiSection_, cvWakeup_, daqDirector_, dataMode_, eventsThisLumi_, Exception, fileDeleteLock_, fileListMode_, fileQueue_, filesToDelete_, fms_, freeChunks_, getNextEventFromDataBlock(), evf::FastMonState::inChunkReceived, evf::FastMonState::inNewLumi, evf::FastMonState::inProcessingFile, evf::FastMonState::inRunEnd, evf::FastMonState::inWaitChunk, evf::FastMonState::inWaitInput, evf::EvFDaqDirector::isSingleStreamThread(), maybeOpenNewLumiSection(), eostools::move(), mWakeup_, evf::EvFDaqDirector::newFile, evf::EvFDaqDirector::newLumi, evf::EvFDaqDirector::noFile, readingFilesCount_, reportEventsThisLumiInSource(), evf::EvFDaqDirector::runAbort, evf::EvFDaqDirector::runEnded, setExceptionState_, setMonState(), mps_update::status, threadError(), and mps_check::timeout.

Referenced by checkNext().

                                                         {
   if (setExceptionState_)
     threadError();
   if (!currentFile_.get()) {
     evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
     setMonState(inWaitInput);
     {
       IdleSourceSentry ids(fms_);
       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("DAQSource::getNextDataBlock") << "Run has been aborted by the input source reader thread";
     } else if (status == evf::EvFDaqDirector::newLumi) {
       setMonState(inNewLumi);
       if (currentFile_->lumi_ > currentLumiSection_) {
         reportEventsThisLumiInSource(currentLumiSection_, eventsThisLumi_);
         eventsThisLumi_ = 0;
         maybeOpenNewLumiSection(currentFile_->lumi_);
       }
       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 (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("DAQSource::getNextDataBlock")
           << "Fully processed " << currentFile_->nProcessed_ << " from the file " << currentFile_->fileName_
           << " but according to BU JSON there should be " << currentFile_->nEvents_ << " events";
     }
     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<RawInputFile>>(currentFileIndex_, std::move(currentFile_)));
     } else {
       //in single-thread and stream jobs, events are already processed
       currentFile_.reset();
     }
     return evf::EvFDaqDirector::noFile;
   }
 
   //assert(currentFile_->status_ == evf::EvFDaqDirector::newFile);
 
   //handle RAW file header
   if (currentFile_->bufferPosition_ == 0 && currentFile_->rawHeaderSize_ > 0) {
     if (currentFile_->fileSize_ <= currentFile_->rawHeaderSize_) {
       if (currentFile_->fileSize_ < currentFile_->rawHeaderSize_)
         throw cms::Exception("DAQSource::getNextDataBlock") << "Premature end of input file while reading file header";
 
       edm::LogWarning("DAQSource") << "File with only raw header and no events received in LS " << currentFile_->lumi_;
       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_->advance(currentFile_->rawHeaderSize_);
   }
 
   //file is too short to fit event header
   if (currentFile_->fileSizeLeft() < dataMode_->headerSize())
     throw cms::Exception("DAQSource::getNextDataBlock")
         << "Premature end of input file while reading event header. Missing: "
         << (dataMode_->headerSize() - currentFile_->fileSizeLeft()) << " bytes";
 
   //multibuffer mode
   //wait for the current chunk to become added to the vector
   setMonState(inWaitChunk);
   {
     IdleSourceSentry ids(fms_);
     while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
       usleep(10000);
       if (setExceptionState_)
         threadError();
     }
   }
   setMonState(inChunkReceived);
 
   chunkIsFree_ = false;
   bool chunkEnd;
   unsigned char* dataPosition;
 
   //read event header, copy it to a single chunk if necessary
   chunkEnd = currentFile_->advance(dataPosition, dataMode_->headerSize());
 
   //get buffer size of current chunk (can be resized)
   uint64_t currentChunkSize = currentFile_->currentChunkSize();
 
   //prepare view based on header that was read
   dataMode_->makeDataBlockView(dataPosition, currentChunkSize, currentFile_->fileSizes_, currentFile_->rawHeaderSize_);
 
   //check that payload size is within the file
   const size_t msgSize = dataMode_->dataBlockSize() - dataMode_->headerSize();
 
   if (currentFile_->fileSizeLeft() < (int64_t)msgSize)
     throw cms::Exception("DAQSource::getNextEventDataBlock")
         << "Premature end of input file (missing:" << (msgSize - currentFile_->fileSizeLeft())
         << ") while parsing block";
 
   //for cross-buffer models
   if (chunkEnd) {
     //header was at the chunk boundary, move payload into the starting chunk as well. No need to update block view here
     currentFile_->moveToPreviousChunk(msgSize, dataMode_->headerSize());
     //mark to release old chunk
     chunkIsFree_ = true;
   } else if (currentChunkSize - currentFile_->chunkPosition_ < msgSize) {
     //header was contiguous, but payload does not fit in the chunk
     //rewind to header start position and then together with payload will be copied together to the old chunk
     currentFile_->rewindChunk(dataMode_->headerSize());
 
     setMonState(inWaitChunk);
     {
       IdleSourceSentry ids(fms_);
       //do the copy to the beginning of the starting chunk. move pointers for next event in the next chunk
       chunkEnd = currentFile_->advance(dataPosition, dataMode_->headerSize() + msgSize);
       assert(chunkEnd);
       //mark to release old chunk
       chunkIsFree_ = true;
     }
     setMonState(inChunkReceived);
     //header and payload is moved, update view
     dataMode_->makeDataBlockView(
         dataPosition, currentFile_->currentChunkSize(), currentFile_->fileSizes_, currentFile_->rawHeaderSize_);
   } else {
     //everything is in a single chunk, only move pointers forward
     chunkEnd = currentFile_->advance(dataPosition, msgSize);
     assert(!chunkEnd);
     chunkIsFree_ = false;
   }
 
   //sanity-check check that the buffer position has not exceeded file size after preparing event
   if (currentFile_->fileSize_ < currentFile_->bufferPosition_)
     throw cms::Exception("DAQSource::getNextEventDataBlock")
         << "Exceeded file size by " << (currentFile_->bufferPosition_ - currentFile_->fileSize_);
 
   //prepare event
   return getNextEventFromDataBlock();
 }

◆ getNextEventFromDataBlock()

evf::EvFDaqDirector::FileStatus DAQSource::getNextEventFromDataBlock ( )

inlineprivate

Definition at line 347 of file DAQSource.cc.

References currentFile_, currentLumiSection_, dataMode_, Exception, fms_, newFWLiteAna::found, evf::FastMonState::inCachedEvent, evf::FastMonState::inChecksumEvent, evf::EvFDaqDirector::noFile, evf::EvFDaqDirector::sameFile, evf::FastMonitoringService::setExceptionDetected(), setMonState(), and verifyChecksum_.

Referenced by checkNext(), and getNextDataBlock().

                                                                  {
   setMonState(inChecksumEvent);
 
   bool found = dataMode_->nextEventView();
   //file(s) completely parsed
   if (!found) {
     if (dataMode_->dataBlockInitialized()) {
       dataMode_->setDataBlockInitialized(false);
       //roll position to the end of the file to close it
       currentFile_->bufferPosition_ = currentFile_->fileSize_;
     }
     return evf::EvFDaqDirector::noFile;
   }
 
   if (verifyChecksum_ && !dataMode_->checksumValid()) {
     if (fms_)
       fms_->setExceptionDetected(currentLumiSection_);
     throw cms::Exception("DAQSource::getNextEventFromDataBlock") << dataMode_->getChecksumError();
   }
   setMonState(inCachedEvent);
 
   currentFile_->nProcessed_++;
 
   return evf::EvFDaqDirector::sameFile;
 }

◆ initFileList()

long DAQSource::initFileList ( )

private

Definition at line 1402 of file DAQSource.cc.

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

Referenced by DAQSource().

                              {
   std::sort(listFileNames_.begin(), listFileNames_.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 (!listFileNames_.empty()) {
     //get run number from first file in the vector
     std::filesystem::path fileName = listFileNames_[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("DAQSource") << "Autodetected run number in fileListMode -: " << rval;
         return rval;
       } catch (const std::exception&) {
         edm::LogWarning("DAQSource") << "Unable to autodetect run number in fileListMode from file -: " << fileName;
       }
     }
   }
   return -1;
 }

◆ makeEventWrapper()

void DAQSource::makeEventWrapper	(	edm::EventPrincipal &	eventPrincipal,
		edm::EventAuxiliary &	aux
	)

inline

Definition at line 54 of file DAQSource.h.

References printConversionInfo::aux, and edm::RawInputSource::makeEvent().

                                                                                    {
     makeEvent(eventPrincipal, aux);
   }

◆ maybeOpenNewLumiSection()

void DAQSource::maybeOpenNewLumiSection ( const uint32_t lumiSection )

private

Definition at line 327 of file DAQSource.cc.

References currentLumiSection_, edm::Timestamp::invalidTimestamp(), edm::InputSource::luminosityBlock(), edm::InputSource::luminosityBlockAuxiliary(), processHistoryID_, edm::InputSource::resetLuminosityBlockAuxiliary(), edm::InputSource::run(), edm::InputSource::runAuxiliary(), and edm::InputSource::setLuminosityBlockAuxiliary().

Referenced by getNextDataBlock().

                                                                   {
   if (!luminosityBlockAuxiliary() || luminosityBlockAuxiliary()->luminosityBlock() != 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("DAQSource") << "New lumi section was opened. LUMI -: " << lumiSection;
   }
 }

◆ processHistoryID()

edm::ProcessHistoryID& DAQSource::processHistoryID ( )

inline

Definition at line 59 of file DAQSource.h.

References processHistoryID_.

59 { return processHistoryID_; }

DAQSource::processHistoryID_

edm::ProcessHistoryID processHistoryID_

Definition: DAQSource.h:117

◆ read()

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

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 548 of file DAQSource.cc.

References checkEvery_, chunkIsFree_, currentFile_, currentFileIndex_, dataMode_, eventsThisLumi_, fileDeleteLock_, filesToDelete_, fms_, freeChunks_, mps_fire::i, evf::FastMonState::inNoRequest, evf::FastMonState::inReadCleanup, evf::FastMonState::inReadEvent, evf::FastMonitoringService::isExceptionOnData(), ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, setMonState(), streamFileTracker_, and edm::EventPrincipal::streamID().

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

                                                       {
   setMonState(inReadEvent);
 
   dataMode_->readEvent(eventPrincipal);
 
   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 < streamFileTracker_.size(); i++) {
         if (it->first == streamFileTracker_.at(i)) {
           fileIsBeingProcessed = true;
           break;
         }
       }
       if (!fileIsBeingProcessed && !(fms_ && fms_->isExceptionOnData(it->second->lumi_))) {
         it = filesToDelete_.erase(it);
       } else
         it++;
     }
   }
   if (dataMode_->dataBlockCompleted() && chunkIsFree_) {
     freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_ - 1]);
     chunkIsFree_ = false;
   }
   setMonState(inNoRequest);
   return;
 }

◆ readSupervisor()

void DAQSource::readSupervisor ( )

private

possibility to use by new formats

Definition at line 593 of file DAQSource.cc.

Referenced by checkNext().

                                {
   bool stop = false;
   unsigned int currentLumiSection = 0;
 
   {
     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.;
 
   bool requireHeader = dataMode_->requireHeader();
 
   while (!stop) {
     //wait for at least one free thread and chunk
     int counter = 0;
 
     while (workerPool_.empty() || 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 % 6000)) {
           edm::LogWarning("FedRawDataInputSource")
               << "No free chunks or threads. Worker pool empty:" << workerPool_.empty()
               << ", free chunks empty:" << freeChunks_.empty() << ", number of files buffered:" << readingFilesCount_
               << " / " << maxBufferedFiles_;
         }
         LogDebug("DAQSource") << "No free chunks or threads...";
       } else {
         assert(!workerPool_.empty() || 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;
     int64_t fileSizeFromMetadata;
 
     if (fms_) {
       setMonStateSup(inSupBusy);
       fms_->startedLookingForFile();
     }
     bool fitToBuffer = dataMode_->fitToBuffer();
 
     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;
 
         unsigned int nConcurrentLumis = daqDirector_->numConcurrentLumis();
         unsigned int nOtherLumis = nConcurrentLumis > 0 ? nConcurrentLumis - 1 : 0;
         unsigned int checkLumiStart = currentLumiSection > nOtherLumis ? currentLumiSection - nOtherLumis : 1;
         bool hasDiscardedLumi = false;
         for (unsigned int i = checkLumiStart; i <= currentLumiSection; i++) {
           if (daqDirector_->lumisectionDiscarded(i)) {
             edm::LogWarning("DAQSource") << "Source detected that the lumisection is discarded -: " << i;
             hasDiscardedLumi = true;
             break;
           }
         }
         if (hasDiscardedLumi)
           break;
 
         setMonStateSup(inThrottled);
         if (!(counter % 50))
           edm::LogWarning("DAQSource") << "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, thisLockWaitTimeUs);
         if (status == evf::EvFDaqDirector::newFile) {
           uint16_t rawDataType;
           if (evf::EvFDaqDirector::parseFRDFileHeader(nextFile,
                                                       rawFd,
                                                       rawHeaderSize,  
                                                       rawDataType,
                                                       lsFromRaw,
                                                       serverEventsInNewFile,
                                                       fileSizeFromMetadata,
                                                       requireHeader,
                                                       false,
                                                       false) != 0) {
             //error
             setExceptionState_ = true;
             stop = true;
             break;
           }
         }
       } else {
         status = daqDirector_->getNextFromFileBroker(currentLumiSection,
                                                      ls,
                                                      nextFile,
                                                      rawFd,
                                                      rawHeaderSize,  //which format?
                                                      serverEventsInNewFile,
                                                      fileSizeFromMetadata,
                                                      thisLockWaitTimeUs,
                                                      requireHeader);
       }
 
       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;
       }
 
       if (status == evf::EvFDaqDirector::runEnded) {
         fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::runEnded));
         stop = true;
         break;
       }
 
       //error from filelocking function
       if (status == evf::EvFDaqDirector::runAbort) {
         fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::runAbort, 0));
         stop = true;
         break;
       }
       //queue new lumisection
       if (ls > currentLumiSection) {
         //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 {
               fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::newLumi, ls));
             }
           } 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++) {
               fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::newLumi, nextLS));
             }
           } else {
             //start from current LS
             fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::newLumi, ls));
           }
         } else {
           //queue all lumisections after last one seen to avoid gaps
           for (unsigned int nextLS = currentLumiSection + 1; nextLS <= ls; nextLS++) {
             fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::newLumi, nextLS));
           }
         }
         currentLumiSection = ls;
       }
       //else
       if (currentLumiSection > 0 && ls < currentLumiSection) {
         edm::LogError("DAQSource") << "Got old LS (" << ls
                                    << ") file from EvFDAQDirector! Expected LS:" << currentLumiSection
                                    << ". Aborting execution." << std::endl;
         if (rawFd != -1)
           close(rawFd);
         rawFd = -1;
         fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::runAbort, 0));
         stop = true;
         break;
       }
 
       int dbgcount = 0;
       if (status == evf::EvFDaqDirector::noFile) {
         setMonStateSup(inSupNoFile);
         dbgcount++;
         if (!(dbgcount % 20))
           LogDebug("DAQSource") << "No file for me... sleep and try again...";
 
         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("DAQSource") << "The director says to grab -: " << nextFile;
 
       std::string rawFile;
       //file service will report raw extension
       rawFile = nextFile;
 
       struct stat st;
       int stat_res = stat(rawFile.c_str(), &st);
       if (stat_res == -1) {
         edm::LogError("DAQSource") << "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 {
         eventsInNewFile = serverEventsInNewFile;
         assert(eventsInNewFile >= 0);
         assert((eventsInNewFile > 0) ==
                (fileSize > rawHeaderSize));  //file without events must be empty or contain only header
       }
 
       std::pair<bool, std::vector<std::string>> additionalFiles =
           dataMode_->defineAdditionalFiles(rawFile, fileListMode_);
       if (!additionalFiles.first) {
         //skip secondary files from file broker
         if (rawFd > -1)
           close(rawFd);
         continue;
       }
 
       std::unique_ptr<RawInputFile> newInputFile(new RawInputFile(evf::EvFDaqDirector::FileStatus::newFile,
                                                                   ls,
                                                                   rawFile,
                                                                   !fileListMode_,
                                                                   rawFd,
                                                                   fileSize,
                                                                   rawHeaderSize,  //for which format
                                                                   0,
                                                                   eventsInNewFile,
                                                                   this));
 
       uint64_t neededSize = fileSize;
       for (const auto& addFile : additionalFiles.second) {
         struct stat buf;
         //wait for secondary files to appear
         unsigned int fcnt = 0;
         while (stat(addFile.c_str(), &buf) != 0) {
           if (fileListMode_) {
             edm::LogError("DAQSource") << "additional file is missing -: " << addFile;
             stop = true;
             setExceptionState_ = true;
             break;
           }
           usleep(10000);
           fcnt++;
           //report and EoR check every 30 seconds
           if ((fcnt && fcnt % 3000 == 0) || quit_threads_.load(std::memory_order_relaxed)) {
             edm::LogWarning("DAQSource") << "Additional file is still missing after 30 seconds -: " << addFile;
             struct stat bufEoR;
             auto secondaryPath = std::filesystem::path(addFile).parent_path();
             auto eorName = std::filesystem::path(daqDirector_->getEoRFileName());
             std::string mainEoR = (std::filesystem::path(daqDirector_->buBaseRunDir()) / eorName).generic_string();
             std::string secondaryEoR = (secondaryPath / eorName).generic_string();
             bool prematureEoR = false;
             if (stat(secondaryEoR.c_str(), &bufEoR) == 0) {
               if (stat(addFile.c_str(), &bufEoR) != 0) {
                 edm::LogError("DAQSource")
                     << "EoR file appeared in -: " << secondaryPath << " while waiting for index file " << addFile;
                 prematureEoR = true;
               }
             } else if (stat(mainEoR.c_str(), &bufEoR) == 0) {
               //wait another 10 seconds
               usleep(10000000);
               if (stat(addFile.c_str(), &bufEoR) != 0) {
                 edm::LogError("DAQSource")
                     << "Main EoR file appeared -: " << mainEoR << " while waiting for index file " << addFile;
                 prematureEoR = true;
               }
             }
             if (prematureEoR) {
               //queue EoR since this is not FU error
               fileQueue_.push(std::make_unique<RawInputFile>(evf::EvFDaqDirector::runEnded, 0));
               stop = true;
               break;
             }
           }
 
           if (quit_threads_) {
             edm::LogError("DAQSource") << "Quitting while waiting for file -: " << addFile;
             stop = true;
             setExceptionState_ = true;
             break;
           }
         }
         LogDebug("DAQSource") << " APPEND NAME " << addFile;
         if (stop)
           break;
 
         newInputFile->appendFile(addFile, buf.st_size);
         neededSize += buf.st_size;
       }
       if (stop)
         break;
 
       //calculate number of needed chunks and size if resizing will be applied
       uint16_t neededChunks;
       uint64_t chunkSize;
 
       if (fitToBuffer) {
         chunkSize = std::min(maxChunkSize_, std::max(eventChunkSize_, neededSize));
         neededChunks = 1;
       } else {
         chunkSize = eventChunkSize_;
         neededChunks = neededSize / eventChunkSize_ + uint16_t((neededSize % eventChunkSize_) > 0);
       }
       newInputFile->setChunks(neededChunks);
 
       newInputFile->randomizeOrder(rng_);
 
       readingFilesCount_++;
       auto newInputFilePtr = newInputFile.get();
       fileQueue_.push(std::move(newInputFile));
 
       for (size_t 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_);
 
         uint64_t toRead = chunkSize;
         if (i == (uint64_t)neededChunks - 1 && neededSize % chunkSize)
           toRead = neededSize % chunkSize;
         newChunk->reset(i * chunkSize, toRead, i);
 
         workerJob_[newTid].first = newInputFilePtr;
         workerJob_[newTid].second = newChunk;
 
         //wake up the worker thread
         cvReader_[newTid]->notify_one();
       }
     }
   }
   setMonStateSup(inRunEnd);
   //make sure threads finish reading
   unsigned int numFinishedThreads = 0;
   while (numFinishedThreads < workerThreads_.size()) {
     unsigned int 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 DAQSource::readWorker ( unsigned int tid )

private

Definition at line 1064 of file DAQSource.cc.

References addFile(), cms::cuda::assert(), InputChunk::buf_, cvReader_, cvWakeup_, dataMode_, change_name::diff, mps_fire::end, eventChunkBlock_, eventChunkSize_, geometryDiff::file, InputChunk::fileIndex_, dqmdumpme::first, mps_fire::i, dqmiolumiharvest::j, dqmdumpme::last, LogDebug, maxChunkSize_, SiStripPI::min, mReader_, submitPVValidationJobs::now, numConcurrentReads_, InputChunk::offset_, read(), InputChunk::readComplete_, InputChunk::resize(), alignCSCRings::s, edm::second(), setExceptionState_, InputChunk::size_, runEdmFileComparison::skipped, command_line::start, startupCv_, startupLock_, AlCaHLTBitMon_QueryRunRegistry::string, thread_quit_signal, threadInit_, tid_active_, InputChunk::usedSize_, workerJob_, and workerPool_.

Referenced by DAQSource().

                                            {
   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> lks(startupLock_);
       init = false;
       startupCv_.notify_one();
     }
     cvWakeup_.notify_all();
     cvReader_[tid]->wait(lk);
     lk.unlock();
 
     if (thread_quit_signal[tid])
       return;
     tid_active_[tid] = true;
 
     RawInputFile* file;
     InputChunk* chunk;
 
     assert(workerJob_[tid].first != nullptr && workerJob_[tid].second != nullptr);
 
     file = workerJob_[tid].first;
     chunk = workerJob_[tid].second;
 
     bool fitToBuffer = dataMode_->fitToBuffer();
 
     //resize if multi-chunked reading is not possible
     if (fitToBuffer) {
       uint64_t accum = 0;
       for (auto s : file->diskFileSizes_)
         accum += s;
       if (accum > eventChunkSize_) {
         if (!chunk->resize(accum, maxChunkSize_)) {
           edm::LogError("DAQSource")
               << "maxChunkSize can not accomodate the file set. Try increasing chunk size and/or chunk maximum size.";
           if (file->rawFd_ != -1 && (numConcurrentReads_ == 1 || chunk->offset_ == 0))
             close(file->rawFd_);
           setExceptionState_ = true;
           continue;
         } else {
           edm::LogInfo("DAQSource") << "chunk size was increased to " << (chunk->size_ >> 20) << " MB";
         }
       }
     }
 
     //skip reading initial header size in first chunk if inheriting file descriptor (already set at appropriate position)
     unsigned int bufferLeftInitial = (chunk->offset_ == 0 && file->rawFd_ != -1) ? file->rawHeaderSize_ : 0;
     const uint16_t readBlocks = chunk->size_ / eventChunkBlock_ + uint16_t(chunk->size_ % eventChunkBlock_ > 0);
 
     auto readPrimary = [&](uint64_t bufferLeft) {
       //BEGIN reading primary file - check if file descriptor is already open
       //in multi-threaded chunked mode, only first thread will use already open fd for reading the first file
       //fd will not be closed in other case (used by other threads)
       int fileDescriptor = -1;
       bool fileOpenedHere = false;
 
       if (numConcurrentReads_ == 1) {
         fileDescriptor = file->rawFd_;
         file->rawFd_ = -1;
         if (fileDescriptor == -1) {
           fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
           fileOpenedHere = true;
         }
       } 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 == -1) {
         edm::LogError("DAQSource") << "readWorker failed to open file -: " << file->fileName_
                                    << " fd:" << fileDescriptor << " error: " << strerror(errno);
         setExceptionState_ = true;
         return;
       }
 
       if (fileOpenedHere) {  //fast forward to this chunk position
         off_t pos = lseek(fileDescriptor, chunk->offset_, SEEK_SET);
         if (pos == -1) {
           edm::LogError("DAQSource") << "readWorker failed to seek file -: " << file->fileName_
                                      << " fd:" << fileDescriptor << " to offset " << chunk->offset_
                                      << " error: " << strerror(errno);
           setExceptionState_ = true;
           return;
         }
       }
 
       LogDebug("DAQSource") << "Reader thread opened file -: TID: " << tid << " file: " << file->fileName_
                             << " at offset " << lseek(fileDescriptor, 0, SEEK_CUR);
 
       size_t skipped = bufferLeft;
       auto start = std::chrono::high_resolution_clock::now();
       for (unsigned int i = 0; i < readBlocks; i++) {
         ssize_t last;
         edm::LogInfo("DAQSource") << "readWorker read -: " << (int64_t)(chunk->usedSize_ - bufferLeft) << " or "
                                   << (int64_t)eventChunkBlock_;
 
         //protect against reading into next block
         last = ::read(fileDescriptor,
                       (void*)(chunk->buf_ + bufferLeft),
                       std::min((int64_t)(chunk->usedSize_ - bufferLeft), (int64_t)eventChunkBlock_));
 
         if (last < 0) {
           edm::LogError("DAQSource") << "readWorker failed to read file -: " << file->fileName_
                                      << " fd:" << fileDescriptor << " last: " << last << " error: " << strerror(errno);
           setExceptionState_ = true;
           break;
         }
         if (last > 0) {
           bufferLeft += last;
         }
         if ((uint64_t)last < eventChunkBlock_) {  //last read
           edm::LogInfo("DAQSource") << "chunkUsedSize" << chunk->usedSize_ << " u-s:" << (chunk->usedSize_ - skipped)
                                     << " ix:" << i * eventChunkBlock_ << " " << (size_t)last;
           //check if this is last block if single file, then total read size must match file size
           if (file->numFiles_ == 1 && !(chunk->usedSize_ - skipped == i * eventChunkBlock_ + (size_t)last)) {
             edm::LogError("DAQSource") << "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_)
         return;
 
       file->fileSizes_[0] = bufferLeft;
 
       if (chunk->offset_ + bufferLeft == file->diskFileSizes_[0] || bufferLeft == chunk->size_) {
         //file reading finished using this fd
         //or the whole buffer is filled (single sequential file spread over more chunks)
         close(fileDescriptor);
         fileDescriptor = -1;
       } else
         assert(fileDescriptor == -1);
 
       if (fitToBuffer && bufferLeft != file->diskFileSizes_[0]) {
         edm::LogError("DAQSource") << "mismatch between read file size for file -: " << file->fileNames_[0]
                                    << " read:" << bufferLeft << " expected:" << file->diskFileSizes_[0];
         setExceptionState_ = true;
         return;
       }
 
       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("DAQSource") << " finished reading block -: " << (bufferLeft >> 20) << " MB"
                             << " in " << msec.count() << " ms (" << (bufferLeft >> 20) / double(msec.count())
                             << " GB/s)";
     };
     //END primary function
 
     //SECONDARY files function
     auto readSecondary = [&](uint64_t bufferLeft, unsigned int j) {
       size_t fileLen = 0;
 
       std::string const& addFile = file->fileNames_[j];
       int fileDescriptor = open(addFile.c_str(), O_RDONLY);
 
       if (fileDescriptor < 0) {
         edm::LogError("DAQSource") << "readWorker failed to open file -: " << addFile << " fd:" << fileDescriptor
                                    << " error: " << strerror(errno);
         setExceptionState_ = true;
         return;
       }
 
       LogDebug("DAQSource") << "Reader thread opened file -: TID: " << tid << " file: " << addFile << " at offset "
                             << lseek(fileDescriptor, 0, SEEK_CUR);
 
       //size_t skipped = 0;//file is newly opened, read with header
       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
         //use bufferLeft for the write offset
         last = ::read(fileDescriptor,
                       (void*)(chunk->buf_ + bufferLeft),
                       std::min((uint64_t)file->diskFileSizes_[j], (uint64_t)eventChunkBlock_));
 
         if (last < 0) {
           edm::LogError("DAQSource") << "readWorker failed to read file -: " << addFile << " fd:" << fileDescriptor
                                      << " error: " << strerror(errno);
           setExceptionState_ = true;
           close(fileDescriptor);
           break;
         }
         if (last > 0) {
           bufferLeft += last;
           fileLen += last;
           file->fileSize_ += last;
         }
       };
 
       close(fileDescriptor);
       file->fileSizes_[j] = fileLen;
       assert(fileLen > 0);
 
       if (fitToBuffer && fileLen != file->diskFileSizes_[j]) {
         edm::LogError("DAQSource") << "mismatch between read file size for file -: " << file->fileNames_[j]
                                    << " read:" << fileLen << " expected:" << file->diskFileSizes_[j];
         setExceptionState_ = true;
         return;
       }
 
       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("DAQSource") << " finished reading block -: " << (bufferLeft >> 20) << " MB"
                             << " in " << msec.count() << " ms (" << (bufferLeft >> 20) / double(msec.count())
                             << " GB/s)";
     };
 
     //randomized order multi-file loop
     for (unsigned int j : file->fileOrder_) {
       if (j == 0) {
         readPrimary(bufferLeftInitial);
       } else
         readSecondary(file->bufferOffsets_[j], j);
 
       if (setExceptionState_)
         break;
     }
 
     if (setExceptionState_)
       continue;
 
     //detect FRD event version. Skip file Header if it exists
     if (dataMode_->dataVersion() == 0 && chunk->offset_ == 0) {
       dataMode_->detectVersion(chunk->buf_, file->rawHeaderSize_);
     }
     assert(dataMode_->versionCheck());
 
     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 DAQSource::reportEventsThisLumiInSource	(	unsigned int	lumi,
		unsigned int	events
	)

private

Definition at line 1381 of file DAQSource.cc.

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

Referenced by checkNext(), and getNextDataBlock().

                                                                                    {
   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 DAQSource::rewind_ ( )

overrideprivatevirtual

Reimplemented from edm::InputSource.

Definition at line 589 of file DAQSource.cc.

589 {}

◆ setMonState()

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

protected

Definition at line 1330 of file DAQSource.cc.

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

Referenced by RawInputFile::advance(), checkNext(), getNextDataBlock(), getNextEventFromDataBlock(), and read().

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

◆ setMonStateSup()

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

protected

Definition at line 1335 of file DAQSource.cc.

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

Referenced by readSupervisor().

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

◆ threadError()

void DAQSource::threadError ( )

private

Definition at line 1325 of file DAQSource.cc.

References Exception, and quit_threads_.

Referenced by RawInputFile::advance(), and getNextDataBlock().

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

◆ useL1EventID()

bool DAQSource::useL1EventID ( ) const

inline

Definition at line 51 of file DAQSource.h.

References useL1EventID_.

51 { return useL1EventID_; }

DAQSource::useL1EventID_

const bool useL1EventID_

Definition: DAQSource.h:103

Friends And Related Function Documentation

◆ InputChunk

friend struct InputChunk

friend

Definition at line 43 of file DAQSource.h.

Referenced by DAQSource().

◆ RawInputFile

friend class RawInputFile

friend

Definition at line 42 of file DAQSource.h.

Referenced by readSupervisor().

Member Data Documentation

◆ alwaysStartFromFirstLS_

const bool DAQSource::alwaysStartFromFirstLS_

private

Definition at line 101 of file DAQSource.h.

Referenced by readSupervisor().

◆ checkEvery_

unsigned int DAQSource::checkEvery_ = 10

private

Definition at line 162 of file DAQSource.h.

Referenced by read().

◆ chunkIsFree_

bool DAQSource::chunkIsFree_ = false

private

Definition at line 135 of file DAQSource.h.

Referenced by getNextDataBlock(), and read().

◆ currentFile_

std::unique_ptr<RawInputFile> DAQSource::currentFile_

private

Definition at line 134 of file DAQSource.h.

Referenced by getNextDataBlock(), getNextEventFromDataBlock(), read(), and ~DAQSource().

◆ currentFileIndex_

int DAQSource::currentFileIndex_ = -1

private

Definition at line 158 of file DAQSource.h.

Referenced by getNextDataBlock(), and read().

◆ currentLumiSection_

unsigned int DAQSource::currentLumiSection_

private

Definition at line 119 of file DAQSource.h.

Referenced by checkNext(), currentLumiSection(), getNextDataBlock(), getNextEventFromDataBlock(), and maybeOpenNewLumiSection().

◆ cvReader_

std::vector<std::unique_ptr<std::condition_variable> > DAQSource::cvReader_

private

Definition at line 149 of file DAQSource.h.

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

◆ cvWakeup_

std::condition_variable DAQSource::cvWakeup_

private

Definition at line 166 of file DAQSource.h.

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

◆ daqDirector_

evf::EvFDaqDirector* DAQSource::daqDirector_ = nullptr

private

Definition at line 88 of file DAQSource.h.

Referenced by checkNext(), DAQSource(), getNextDataBlock(), and readSupervisor().

◆ dataArrangerThread_

std::unique_ptr<std::thread> DAQSource::dataArrangerThread_

private

Definition at line 139 of file DAQSource.h.

◆ dataMode_

std::shared_ptr<DataMode> DAQSource::dataMode_

private

Definition at line 176 of file DAQSource.h.

Referenced by checkNext(), DAQSource(), getNextDataBlock(), getNextEventFromDataBlock(), read(), readSupervisor(), and readWorker().

◆ dataModeConfig_

const std::string DAQSource::dataModeConfig_

private

Definition at line 90 of file DAQSource.h.

Referenced by DAQSource().

◆ eventChunkBlock_

uint64_t DAQSource::eventChunkBlock_

private

Definition at line 93 of file DAQSource.h.

Referenced by DAQSource(), and readWorker().

◆ eventChunkSize_

uint64_t DAQSource::eventChunkSize_

private

Definition at line 91 of file DAQSource.h.

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

◆ eventRunNumber_

uint32_t DAQSource::eventRunNumber_ = 0

private

Definition at line 120 of file DAQSource.h.

Referenced by checkNext(), and eventRunNumber().

◆ eventsThisLumi_

unsigned int DAQSource::eventsThisLumi_

private

Definition at line 122 of file DAQSource.h.

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

◆ eventsThisRun_

unsigned long DAQSource::eventsThisRun_ = 0

private

Definition at line 123 of file DAQSource.h.

◆ fileDeleteLock_

std::mutex DAQSource::fileDeleteLock_

private

Definition at line 160 of file DAQSource.h.

Referenced by DAQSource(), getNextDataBlock(), read(), and ~DAQSource().

◆ fileDescriptor_

int DAQSource::fileDescriptor_ = -1

private

Definition at line 169 of file DAQSource.h.

◆ fileListIndex_

unsigned int DAQSource::fileListIndex_ = 0

private

Definition at line 110 of file DAQSource.h.

Referenced by getFile().

◆ fileListLoopMode_

const bool DAQSource::fileListLoopMode_

private

Definition at line 111 of file DAQSource.h.

Referenced by checkNext(), DAQSource(), fileListLoopMode(), and getFile().

◆ fileListMode_

const bool DAQSource::fileListMode_

private

Definition at line 109 of file DAQSource.h.

Referenced by checkNext(), DAQSource(), getNextDataBlock(), and readSupervisor().

◆ fileQueue_

tbb::concurrent_queue<std::unique_ptr<RawInputFile> > DAQSource::fileQueue_

private

Definition at line 146 of file DAQSource.h.

Referenced by getNextDataBlock(), and readSupervisor().

◆ filesToDelete_

std::list<std::pair<int, std::unique_ptr<InputFile> > > DAQSource::filesToDelete_

private

Definition at line 159 of file DAQSource.h.

Referenced by DAQSource(), getNextDataBlock(), read(), and ~DAQSource().

◆ fms_

evf::FastMonitoringService* DAQSource::fms_ = nullptr

private

Definition at line 87 of file DAQSource.h.

Referenced by DAQSource(), getNextDataBlock(), getNextEventFromDataBlock(), read(), readSupervisor(), setMonState(), setMonStateSup(), and ~DAQSource().

◆ freeChunks_

tbb::concurrent_queue<InputChunk*> DAQSource::freeChunks_

private

Definition at line 145 of file DAQSource.h.

Referenced by DAQSource(), getNextDataBlock(), read(), and readSupervisor().

◆ fuOutputDir_

std::string DAQSource::fuOutputDir_

private

Definition at line 115 of file DAQSource.h.

◆ GTPEventID_

uint32_t DAQSource::GTPEventID_ = 0

private

Definition at line 121 of file DAQSource.h.

◆ listFileNames_

std::vector<std::string> DAQSource::listFileNames_

private

Definition at line 105 of file DAQSource.h.

Referenced by getFile(), and initFileList().

◆ loopModeIterationInc_

unsigned int DAQSource::loopModeIterationInc_ = 0

private

Definition at line 112 of file DAQSource.h.

Referenced by getFile().

◆ maxBufferedFiles_

unsigned int DAQSource::maxBufferedFiles_

private

Definition at line 96 of file DAQSource.h.

Referenced by readSupervisor().

◆ maxChunkSize_

uint64_t DAQSource::maxChunkSize_

private

Definition at line 92 of file DAQSource.h.

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

◆ monlock_

std::mutex DAQSource::monlock_

private

Definition at line 174 of file DAQSource.h.

Referenced by getEventReport(), and reportEventsThisLumiInSource().

◆ mReader_

std::mutex DAQSource::mReader_

private

Definition at line 148 of file DAQSource.h.

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

◆ mWakeup_

std::mutex DAQSource::mWakeup_

private

Definition at line 165 of file DAQSource.h.

Referenced by getNextDataBlock(), and readSupervisor().

◆ numBuffers_

unsigned int DAQSource::numBuffers_

private

Definition at line 95 of file DAQSource.h.

Referenced by DAQSource().

◆ numConcurrentReads_

unsigned int DAQSource::numConcurrentReads_

private

Definition at line 97 of file DAQSource.h.

Referenced by DAQSource(), and readWorker().

◆ processHistoryID_

edm::ProcessHistoryID DAQSource::processHistoryID_

private

Definition at line 117 of file DAQSource.h.

Referenced by DAQSource(), maybeOpenNewLumiSection(), and processHistoryID().

◆ quit_threads_

std::atomic<bool> DAQSource::quit_threads_

private

Definition at line 152 of file DAQSource.h.

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

◆ readBlocks_

unsigned int DAQSource::readBlocks_

private

Definition at line 94 of file DAQSource.h.

Referenced by DAQSource().

◆ readingFilesCount_

std::atomic<unsigned int> DAQSource::readingFilesCount_

private

Definition at line 98 of file DAQSource.h.

Referenced by DAQSource(), getNextDataBlock(), and readSupervisor().

◆ readSupervisorThread_

std::unique_ptr<std::thread> DAQSource::readSupervisorThread_

private

Definition at line 138 of file DAQSource.h.

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

◆ rng_

std::default_random_engine DAQSource::rng_

private

Definition at line 124 of file DAQSource.h.

Referenced by readSupervisor().

◆ runNumber_

edm::RunNumber_t DAQSource::runNumber_

private

Definition at line 114 of file DAQSource.h.

Referenced by checkNext(), and DAQSource().

◆ setExceptionState_

bool DAQSource::setExceptionState_ = false

private

Definition at line 154 of file DAQSource.h.

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

◆ sourceEventsReport_

std::map<unsigned int, unsigned int> DAQSource::sourceEventsReport_

private

Definition at line 173 of file DAQSource.h.

Referenced by getEventReport(), and reportEventsThisLumiInSource().

◆ startedSupervisorThread_

bool DAQSource::startedSupervisorThread_ = false

private

Definition at line 137 of file DAQSource.h.

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

◆ startupCv_

std::condition_variable DAQSource::startupCv_

private

Definition at line 156 of file DAQSource.h.

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

◆ startupLock_

std::mutex DAQSource::startupLock_

private

Definition at line 155 of file DAQSource.h.

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

◆ streamFileTracker_

std::vector<int> DAQSource::streamFileTracker_

private

Definition at line 161 of file DAQSource.h.

Referenced by read().

◆ testTCDSFEDRange_

const std::vector<unsigned int> DAQSource::testTCDSFEDRange_

private

Definition at line 104 of file DAQSource.h.

Referenced by DAQSource().

◆ thread_quit_signal

std::vector<unsigned int> DAQSource::thread_quit_signal

private

Definition at line 153 of file DAQSource.h.

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

◆ threadInit_

std::atomic<bool> DAQSource::threadInit_

private

Definition at line 171 of file DAQSource.h.

Referenced by readWorker().

◆ tid_active_

std::vector<unsigned int> DAQSource::tid_active_

private

Definition at line 150 of file DAQSource.h.

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

◆ useFileBroker_

bool DAQSource::useFileBroker_

private

Definition at line 106 of file DAQSource.h.

◆ useL1EventID_

const bool DAQSource::useL1EventID_

private

Definition at line 103 of file DAQSource.h.

Referenced by useL1EventID().

◆ verifyChecksum_

const bool DAQSource::verifyChecksum_

private

Definition at line 102 of file DAQSource.h.

Referenced by getNextEventFromDataBlock().

◆ workerJob_

std::vector<ReaderInfo> DAQSource::workerJob_

private

Definition at line 143 of file DAQSource.h.

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

◆ workerPool_

tbb::concurrent_queue<unsigned int> DAQSource::workerPool_

private

Definition at line 142 of file DAQSource.h.

Referenced by readSupervisor(), and readWorker().

◆ workerThreads_

std::vector<std::thread*> DAQSource::workerThreads_

private

Definition at line 140 of file DAQSource.h.

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

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

◆ DAQSource()

◆ ~DAQSource()

Member Function Documentation

◆ checkNext()

◆ currentLumiSection()

◆ dataArranger()

◆ eventRunNumber()

◆ exceptionState()

◆ fileListLoopMode()

◆ fillDescriptions()

◆ getEventReport()

◆ getFile()

◆ getNextDataBlock()

◆ getNextEventFromDataBlock()

◆ initFileList()

◆ makeEventWrapper()

◆ maybeOpenNewLumiSection()

◆ processHistoryID()

◆ read()

◆ readSupervisor()

◆ readWorker()

◆ reportEventsThisLumiInSource()

◆ rewind_()

◆ setMonState()

◆ setMonStateSup()

◆ threadError()

◆ useL1EventID()

Friends And Related Function Documentation

◆ InputChunk

◆ RawInputFile

Member Data Documentation

◆ alwaysStartFromFirstLS_

◆ checkEvery_

◆ chunkIsFree_

◆ currentFile_

◆ currentFileIndex_

◆ currentLumiSection_

◆ cvReader_

◆ cvWakeup_

◆ daqDirector_

◆ dataArrangerThread_

◆ dataMode_

◆ dataModeConfig_

◆ eventChunkBlock_

◆ eventChunkSize_

◆ eventRunNumber_

◆ eventsThisLumi_

◆ eventsThisRun_

◆ fileDeleteLock_

◆ fileDescriptor_

◆ fileListIndex_

◆ fileListLoopMode_

◆ fileListMode_

◆ fileQueue_

◆ filesToDelete_

◆ fms_

◆ freeChunks_

◆ fuOutputDir_

◆ GTPEventID_

◆ listFileNames_

◆ loopModeIterationInc_

◆ maxBufferedFiles_

◆ maxChunkSize_

◆ monlock_

◆ mReader_

◆ mWakeup_

◆ numBuffers_

◆ numConcurrentReads_