#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)

virtual	~FedRawDataInputSource ()

Public Member Functions inherited from edm::RawInputSource
	RawInputSource (ParameterSet const &pset, InputSourceDescription const &desc)

virtual	~RawInputSource ()

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

void	doBeginLumi (LuminosityBlockPrincipal &lbp, ProcessContext const *)
	Called by framework at beginning of lumi block. More...

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	doEndLumi (LuminosityBlockPrincipal &lbp, bool cleaningUpAfterException, ProcessContext const *)
	Called by framework at end of lumi block. More...

void	doEndRun (RunPrincipal &rp, bool cleaningUpAfterException, ProcessContext const *)
	Called by framework at end of run. More...

void	doPostForkReacquireResources (std::shared_ptr< multicore::MessageReceiverForSource >)

void	doPreForkReleaseResources ()
	Called by the framework before forking the process. 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)
	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...

InputSource &	operator= (InputSource const &)=delete

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::unique_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	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

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

SharedResourcesAcquirer *	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)

bool	skipForForking ()

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 ()
	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
virtual bool	checkNextEvent () override

virtual void	read (edm::EventPrincipal &eventPrincipal) override

Protected Member Functions inherited from edm::RawInputSource
void	makeEvent (EventPrincipal &eventPrincipal, EventAuxiliary const &eventAuxiliary)

void	setInputFileTransitionsEachEvent ()

Protected Member Functions inherited from edm::InputSource
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
void	createBoLSFile (const uint32_t lumiSection, bool checkIfExists)

void	deleteFile (std::string const &)

bool	exceptionState ()

edm::Timestamp	fillFEDRawDataCollection (FEDRawDataCollection &)

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

evf::EvFDaqDirector::FileStatus	getNextEvent ()

int	grabNextJsonFile (boost::filesystem::path const &)

long	initFileList ()

void	maybeOpenNewLumiSection (const uint32_t lumiSection)

evf::EvFDaqDirector::FileStatus	nextEvent ()

virtual void	postForkReacquireResources (std::shared_ptr< edm::multicore::MessageReceiverForSource >) override

virtual void	preForkReleaseResources () override

void	readNextChunkIntoBuffer (InputFile *file)

void	readSupervisor ()

void	readWorker (unsigned int tid)

void	reportEventsThisLumiInSource (unsigned int lumi, unsigned int events)

virtual void	rewind_ () override

void	threadError ()

Private Attributes
uint32_t	bufferInputRead_ = 0

unsigned int	checkEvery_ = 10

bool	chunkIsFree_ =false

InputFile *	currentFile_ = 0

int	currentFileIndex_ = -1

unsigned int	currentLumiSection_

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

std::condition_variable	cvWakeup_

evf::EvFDaqDirector *	daqDirector_ = 0

const edm::DaqProvenanceHelper	daqProvenanceHelper_

std::string	defPath_

uint32	detectedFRDversion_ =0

jsoncollector::DataPointDefinition *	dpd_

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 < InputFile * >	fileQueue_

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

evf::FastMonitoringService *	fms_ = 0

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_

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 > *	streamFileTrackerPtr_ = 0

unsigned char *	tcds_pointer_

std::vector< unsigned int >	thread_quit_signal

std::atomic< bool >	threadInit_

std::vector< unsigned int >	tid_active_

const bool	useL1EventID_

const bool	verifyAdler32_

const bool	verifyChecksum_

std::vector< ReaderInfo >	workerJob_

tbb::concurrent_queue < unsigned int >	workerPool_

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

Friends
class	InputChunk

class	InputFile
	Open Root file and provide MEs ############. More...

Additional Inherited Members
Public Types inherited from edm::InputSource
enum	ItemType { IsInvalid, IsStop, IsFile, IsRun, IsLumi, IsEvent, IsRepeat, IsSynchronize }

enum	ProcessingMode { Runs, RunsAndLumis, RunsLumisAndEvents }

typedef ProductRegistryHelper::TypeLabelList	TypeLabelList

Detailed Description

Definition at line 43 of file FedRawDataInputSource.h.

Member Typedef Documentation

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

private

Definition at line 138 of file FedRawDataInputSource.h.

Constructor & Destructor Documentation

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

explicit

Definition at line 54 of file FedRawDataInputSource.cc.

References assert(), edm::Timestamp::beginOfTime(), crc32c_hw_test(), cvReader_, daqDirector_, edm::DaqProvenanceHelper::daqInit(), daqProvenanceHelper_, defPath_, dpd_, eventChunkBlock_, eventChunkSize_, Exception, fileDeleteLock_, fileListLoopMode_, fileListMode_, filesToDelete_, fms_, freeChunks_, i, evf::FastMonitoringThread::inInit, initFileList(), InputChunk, edm::Timestamp::invalidTimestamp(), numBuffers_, numConcurrentReads_, cppFunctionSkipper::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_, AlCaHLTBitMon_QueryRunRegistry::string, thread_quit_signal, threadInit_, tid_active_, workerJob_, and workerThreads_.

                                                                                     :
   edm::RawInputSource(pset, desc),
   defPath_(pset.getUntrackedParameter<std::string> ("buDefPath", std::string(getenv("CMSSW_BASE"))+"/src/EventFilter/Utilities/plugins/budef.jsd")),
   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)),
   verifyAdler32_(pset.getUntrackedParameter<bool> ("verifyAdler32", true)),
   verifyChecksum_(pset.getUntrackedParameter<bool> ("verifyChecksum", true)),
   useL1EventID_(pset.getUntrackedParameter<bool> ("useL1EventID", false)),
   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()),
   fuOutputDir_(std::string()),
   daqProvenanceHelper_(edm::TypeID(typeid(FEDRawDataCollection))),
   eventID_(),
   processHistoryID_(),
   currentLumiSection_(0),
   tcds_pointer_(0),
   eventsThisLumi_(0),
   dpd_(nullptr)
 {
   char thishost[256];
   gethostname(thishost, 255);
   edm::LogInfo("FedRawDataInputSource") << "Construction. read-ahead chunk size -: "
                                         << std::endl << (eventChunkSize_/1048576)
                                         << " MB on host " << thishost;
 
   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()));
 
   std::string defPathSuffix = "src/EventFilter/Utilities/plugins/budef.jsd";
   defPath_ = std::string(getenv("CMSSW_BASE")) + "/" + defPathSuffix;
   struct stat statbuf;
   if (stat(defPath_.c_str(), &statbuf)) {
     defPath_ = std::string(getenv("CMSSW_RELEASE_BASE")) + "/" + defPathSuffix;
     if (stat(defPath_.c_str(), &statbuf)) {
       defPath_ = defPathSuffix;
     }
   }
 
   dpd_ = new DataPointDefinition();
   std::string defLabel = "data";
   DataPointDefinition::getDataPointDefinitionFor(defPath_, dpd_,&defLabel);
 
   //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(...) {}
   }
   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";
 
   //set DaqDirector to delete files in preGlobalEndLumi callback
   daqDirector_->setDeleteTracking(&fileDeleteLock_,&filesToDelete_);
   if (fms_) {
    daqDirector_->setFMS(fms_);
    fms_->setInputSource(this);
    fms_->setInState(evf::FastMonitoringThread::inInit);
    fms_->setInStateSup(evf::FastMonitoringThread::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 ( )

virtual

Definition at line 182 of file FedRawDataInputSource.cc.

References cvReader_, deleteFile(), filesToDelete_, i, mReader_, numConcurrentReads_, quit_threads_, readSupervisorThread_, startedSupervisorThread_, thread_quit_signal, and workerThreads_.

 {
   quit_threads_=true;
 
   //delete any remaining open files
   for (auto it = filesToDelete_.begin();it!=filesToDelete_.end();it++) {
     deleteFile(it->second->fileName_);
     delete it->second;
   }
   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

bool FedRawDataInputSource::checkNextEvent ( )

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 232 of file FedRawDataInputSource.cc.

 {
   if (!startedSupervisorThread_)
   {
     //late init of directory variable
     fuOutputDir_=edm::Service<evf::EvFDaqDirector>()->baseRunDir();
 
     //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_.reset(new std::thread(&FedRawDataInputSource::readSupervisor,this));
     startedSupervisorThread_=true;
     startupCv_.wait(lk);
   }
   //signal hltd to start event accounting
   if (!currentLumiSection_ && daqDirector_->emptyLumisectionMode())
     daqDirector_->createProcessingNotificationMaybe();
   if (fms_) fms_->setInState(evf::FastMonitoringThread::inWaitInput);
   switch (nextEvent() ) {
     case evf::EvFDaqDirector::runEnded: {
       //maybe create EoL file in working directory before ending run
       struct stat buf;
       if ( 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 false;
     }
     case evf::EvFDaqDirector::noFile: {
       //this is not reachable
       return true;
     }
     case evf::EvFDaqDirector::newLumi: {
       //std::cout << "--------------NEW LUMI---------------" << std::endl;
       return true;
     }
     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 true;
     }
   }
 }

void FedRawDataInputSource::createBoLSFile	(	const uint32_t	lumiSection,
		bool	checkIfExists
	)

private

Definition at line 309 of file FedRawDataInputSource.cc.

References daqDirector_, evf::EvFDaqDirector::getBoLSFilePathOnFU(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by maybeOpenNewLumiSection().

 {
   //used for backpressure mechanisms and monitoring
   const std::string fuBoLS = daqDirector_->getBoLSFilePathOnFU(lumiSection);
   struct stat buf;
   if (checkIfExists==false || stat(fuBoLS.c_str(), &buf) != 0) {
     int bol_fd = open(fuBoLS.c_str(), O_RDWR|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH);
     close(bol_fd);
   }
 }

void FedRawDataInputSource::deleteFile ( std::string const & fileName )

private

Definition at line 641 of file FedRawDataInputSource.cc.

References cppFunctionSkipper::exception, fileListMode_, MillePedeFileConverter_cfg::fileName, LogDebug, fed_dqm_sourceclient-live_cfg::path, and MatrixUtil::remove().

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

 {
   //no deletion in file list mode
   if (fileListMode_) return;
 
   const boost::filesystem::path filePath(fileName);
   LogDebug("FedRawDataInputSource") << "Deleting input file -:" << fileName;
   try {
     //sometimes this fails but file gets deleted
     boost::filesystem::remove(filePath);
   }
   catch (const boost::filesystem::filesystem_error& ex)
   {
     edm::LogError("FedRawDataInputSource") << " - deleteFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what()
                                            << ". Trying again.";
     usleep(100000);
     try {
       boost::filesystem::remove(filePath);
     }
     catch (const boost::filesystem::filesystem_error&) {/*file gets deleted first time but exception is still thrown*/}
   }
   catch (std::exception& ex)
   {
     edm::LogError("FedRawDataInputSource") << " - deleteFile std::exception CAUGHT -: " << ex.what()
                                            << ". Trying again.";
     usleep(100000);
     try {
       boost::filesystem::remove(filePath);
     } catch (std::exception&) {/*file gets deleted first time but exception is still thrown*/}
   }
 }

bool FedRawDataInputSource::exceptionState ( )

inlineprivate

Definition at line 74 of file FedRawDataInputSource.h.

References setExceptionState_.

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

74 {return setExceptionState_;}

FedRawDataInputSource::setExceptionState_

bool setExceptionState_

Definition: FedRawDataInputSource.h:160

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

static

Definition at line 215 of file FedRawDataInputSource.cc.

References edm::ConfigurationDescriptions::add(), edm::ParameterSetDescription::addUntracked(), edm::ParameterSetDescription::setAllowAnything(), edm::ParameterSetDescription::setComment(), and edm::ParameterDescriptionNode::setComment().

 {
   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<bool> ("verifyAdler32", true)->setComment("Verify event Adler32 checksum with FRDv3 or v4");
   desc.addUntracked<bool> ("verifyChecksum", true)->setComment("Verify event CRC-32C checksum of FRDv5 or higher");
   desc.addUntracked<bool> ("useL1EventID", false)->setComment("Use L1 event ID from FED header if true or from TCDS FED if false");
   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);
 }

edm::Timestamp FedRawDataInputSource::fillFEDRawDataCollection ( FEDRawDataCollection & rawData )

private

Definition at line 746 of file FedRawDataInputSource.cc.

References assert(), FEDRawData::data(), event(), event_, fedt_struct::eventsize, evf::evtn::evm_board_sense(), Exception, FED_EVSZ_EXTRACT, FED_SOID_EXTRACT, FEDRawDataCollection::FEDData(), l1tstage2_dqm_sourceclient-live_cfg::fedId, evf::evtn::get(), evf::evtn::getgpshigh(), evf::evtn::getgpslow(), evf::evtn::gtpe_board_sense(), evf::evtn::gtpe_get(), GTPEventID_, FEDNumbering::MAXFEDID, FEDNumbering::MINTCDSuTCAFEDID, FEDNumbering::MINTriggerEGTPFEDID, FEDNumbering::MINTriggerGTPFEDID, FEDRawData::resize(), fedh_struct::sourceid, and tcds_pointer_.

Referenced by read().

 {
   edm::TimeValue_t time;
   timeval stv;
   gettimeofday(&stv,0);
   time = stv.tv_sec;
   time = (time << 32) + stv.tv_usec;
   edm::Timestamp tstamp(time);
 
   uint32_t eventSize = event_->eventSize();
   char* event = (char*)event_->payload();
   GTPEventID_=0;
   tcds_pointer_ = 0;
   while (eventSize > 0) {
     assert(eventSize>=sizeof(fedt_t));
     eventSize -= sizeof(fedt_t);
     const fedt_t* fedTrailer = (fedt_t*) (event + eventSize);
     const uint32_t fedSize = FED_EVSZ_EXTRACT(fedTrailer->eventsize) << 3; //trailer length counts in 8 bytes
     assert(eventSize>=fedSize - sizeof(fedt_t));
     eventSize -= (fedSize - sizeof(fedt_t));
     const fedh_t* fedHeader = (fedh_t *) (event + eventSize);
     const uint16_t fedId = FED_SOID_EXTRACT(fedHeader->sourceid);
     if(fedId>FEDNumbering::MAXFEDID)
     {
       throw cms::Exception("FedRawDataInputSource::fillFEDRawDataCollection") << "Out of range FED ID : " << fedId;
     }
     if (fedId == FEDNumbering::MINTCDSuTCAFEDID) {
       tcds_pointer_ = (unsigned char *)(event + eventSize );
     }
     if (fedId == FEDNumbering::MINTriggerGTPFEDID) {
       if (evf::evtn::evm_board_sense((unsigned char*) fedHeader,fedSize))
           GTPEventID_ = evf::evtn::get((unsigned char*) fedHeader,true);
       else
           GTPEventID_ = evf::evtn::get((unsigned char*) fedHeader,false);
       //evf::evtn::evm_board_setformat(fedSize);
       const uint64_t gpsl = evf::evtn::getgpslow((unsigned char*) fedHeader);
       const uint64_t gpsh = evf::evtn::getgpshigh((unsigned char*) fedHeader);
       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((unsigned char*)fedHeader)) {
         GTPEventID_ = evf::evtn::gtpe_get((unsigned char*) fedHeader);
       }
     }
     FEDRawData& fedData = rawData.FEDData(fedId);
     fedData.resize(fedSize);
     memcpy(fedData.data(), event + eventSize, fedSize);
   }
   assert(eventSize == 0);
 
   return tstamp;
 }

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

Definition at line 1403 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()) {
     auto && ret = std::pair<bool,unsigned int>(true,itr->second);
     if (erase) 
       sourceEventsReport_.erase(itr);
     return ret;
   }
   else 
     return std::pair<bool,unsigned int>(false,0);
 }

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

private

Definition at line 1446 of file FedRawDataInputSource.cc.

References fileListIndex_, fileListLoopMode_, MillePedeFileConverter_cfg::fileName, fileNames_, loopModeIterationInc_, evf::EvFDaqDirector::newFile, fed_dqm_sourceclient-live_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);
           boost::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 = boost::lexical_cast<unsigned int>(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);
     }
   }
 }

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

inlineprivate

Definition at line 371 of file FedRawDataInputSource.cc.

Referenced by nextEvent().

 {
 
   if (setExceptionState_) threadError();
   if (!currentFile_)
   {
     if (!streamFileTrackerPtr_) {
       streamFileTrackerPtr_ = daqDirector_->getStreamFileTracker();
       nStreams_ = streamFileTrackerPtr_->size();
       if (nStreams_>10) checkEvery_=nStreams_;
     }
 
     evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
     if (fms_) fms_->setInState(evf::FastMonitoringThread::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_)
         return evf::EvFDaqDirector::noFile;
     }
     status = currentFile_->status_;
     if ( status == evf::EvFDaqDirector::runEnded)
     {
       if (fms_) fms_->setInState(evf::FastMonitoringThread::inRunEnd);
       delete currentFile_;
       currentFile_=nullptr;
       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)
     {
       if (fms_) fms_->setInState(evf::FastMonitoringThread::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;
       }
 
       delete currentFile_;
       currentFile_=nullptr;
       return status;
     }
     else if (status == evf::EvFDaqDirector::newFile) {
       currentFileIndex_++;
       daqDirector_->updateFileIndex(currentFileIndex_);
     }
     else
       assert(0);
   }
   if (fms_) fms_->setInState(evf::FastMonitoringThread::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
     deleteFile(currentFile_->fileName_);
     delete currentFile_;
     currentFile_=nullptr;
     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,InputFile*>(currentFileIndex_,currentFile_));
     }
     else {
       //in single-thread and stream jobs, events are already processed
       deleteFile(currentFile_->fileName_);
       delete currentFile_;
     }
     currentFile_=nullptr;
     return evf::EvFDaqDirector::noFile;
   }
 
 
   //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
     if (fms_) fms_->setInState(evf::FastMonitoringThread::inWaitChunk);
     while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
       usleep(10000);
       if (currentFile_->parent_->exceptionState()) currentFile_->parent_->threadError();
     }
     if (fms_) fms_->setInState(evf::FastMonitoringThread::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_);
 
       if (detectedFRDversion_==0) {
         detectedFRDversion_=*((uint32*)dataPosition);
         if (detectedFRDversion_>5) 
           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_.reset( new 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_);
       //recalculate chunk position
       dataPosition = currentFile_->chunks_[0]->buf_+ currentFile_->chunkPosition_;
       event_.reset( new 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
     if (fms_) fms_->setInState(evf::FastMonitoringThread::inWaitChunk);
     while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
       usleep(10000);
       if (setExceptionState_) threadError();
     }
     if (fms_) fms_->setInState(evf::FastMonitoringThread::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_.reset( new 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
     chunkEnd = currentFile_->advance(dataPosition,FRDHeaderVersionSize[detectedFRDversion_]+msgSize);
     assert(chunkEnd);
     chunkIsFree_=true;
     //header is moved
     event_.reset( new 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
   if (fms_) fms_->setInState(evf::FastMonitoringThread::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 ( verifyAdler32_ && 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;
     }
   }
   if (fms_) fms_->setInState(evf::FastMonitoringThread::inCachedEvent);
 
   currentFile_->nProcessed_++;
 
   return evf::EvFDaqDirector::sameFile;
 }

int FedRawDataInputSource::grabNextJsonFile ( boost::filesystem::path const & jsonSourcePath )

private

Definition at line 800 of file FedRawDataInputSource.cc.

References filterCSVwithJSON::copy, daqDirector_, data, jsoncollector::DataPoint::deserialize(), reco::dp, dpd_, alignCSCRings::e, cppFunctionSkipper::exception, Exception, fuOutputDir_, jsoncollector::DataPoint::getData(), Json::Reader::getFormatedErrorMessages(), jsoncollector::DataPointDefinition::getNames(), i, LogDebug, Json::Reader::parse(), fed_dqm_sourceclient-live_cfg::path, matplotRender::reader, MatrixUtil::remove(), contentValuesCheck::ss, AlCaHLTBitMon_QueryRunRegistry::string, summarizeEdmComparisonLogfiles::success, and evf::EvFDaqDirector::unlockFULocal().

Referenced by readSupervisor().

 {
   std::string data;
   try {
     // assemble json destination path
     boost::filesystem::path jsonDestPath(fuOutputDir_);
 
     //TODO:should be ported to use fffnaming
     std::ostringstream fileNameWithPID;
     fileNameWithPID << jsonSourcePath.stem().string() << "_pid"
                     << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
     jsonDestPath /= fileNameWithPID.str();
 
     LogDebug("FedRawDataInputSource") << "JSON rename -: " << jsonSourcePath << " to "
                                           << jsonDestPath;
     try {
       boost::filesystem::copy(jsonSourcePath,jsonDestPath);
     }
     catch (const boost::filesystem::filesystem_error& ex)
     {
       // Input dir gone?
       edm::LogError("FedRawDataInputSource") << "grabNextFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what();
       //                                     << " Maybe the file is not yet visible by FU. Trying again in one second";
       sleep(1);
       boost::filesystem::copy(jsonSourcePath,jsonDestPath);
     }
     daqDirector_->unlockFULocal();
 
     try {
       //sometimes this fails but file gets deleted
       boost::filesystem::remove(jsonSourcePath);
     }
     catch (const boost::filesystem::filesystem_error& ex)
     {
       // Input dir gone?
       edm::LogError("FedRawDataInputSource") << "grabNextFile BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what();
     }
     catch (std::exception& ex)
     {
       // Input dir gone?
       edm::LogError("FedRawDataInputSource") << "grabNextFile std::exception CAUGHT -: " << ex.what();
     }
 
     boost::filesystem::ifstream ij(jsonDestPath);
     Json::Value deserializeRoot;
     Json::Reader reader;
 
     std::stringstream ss;
     ss << ij.rdbuf();
     if (!reader.parse(ss.str(), deserializeRoot)) {
       edm::LogError("FedRawDataInputSource") << "Failed to deserialize JSON file -: " << jsonDestPath
                                                << "\nERROR:\n" << reader.getFormatedErrorMessages()
                                                << "CONTENT:\n" << ss.str()<<".";
       throw std::runtime_error("Cannot deserialize input JSON file");
     }
 
     //read BU JSON
     std::string data;
     DataPoint dp;
     dp.deserialize(deserializeRoot);
     bool success = false;
     for (unsigned int i=0;i<dpd_->getNames().size();i++) {
       if (dpd_->getNames().at(i)=="NEvents")
     if (i<dp.getData().size()) {
       data = dp.getData()[i];
       success=true;
     }
     }
     if (!success) {
       if (dp.getData().size())
     data = dp.getData()[0];
       else
     throw cms::Exception("FedRawDataInputSource::grabNextJsonFile") <<
       " error reading number of events from BU JSON -: No input value " << data;
     }
     return boost::lexical_cast<int>(data);
 
   }
   catch (const boost::filesystem::filesystem_error& ex)
   {
     // Input dir gone?
     daqDirector_->unlockFULocal();
     edm::LogError("FedRawDataInputSource") << "grabNextJsonFile - BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what();
   }
   catch (std::runtime_error e)
   {
     // Another process grabbed the file and NFS did not register this
     daqDirector_->unlockFULocal();
     edm::LogError("FedRawDataInputSource") << "grabNextJsonFile - runtime Exception -: " << e.what();
   }
 
   catch( boost::bad_lexical_cast const& ) {
     edm::LogError("FedRawDataInputSource") << "grabNextJsonFile - error parsing number of events from BU JSON. "
                                            << "Input value is -: " << data;
   }
 
   catch (std::exception e)
   {
     // BU run directory disappeared?
     daqDirector_->unlockFULocal();
     edm::LogError("FedRawDataInputSource") << "grabNextFile - SOME OTHER EXCEPTION OCCURED!!!! -: " << e.what();
   }
 
   return -1;
 }

long FedRawDataInputSource::initFileList ( )

private

Definition at line 1417 of file FedRawDataInputSource.cc.

References a, b, end, MillePedeFileConverter_cfg::fileName, fileNames_, fed_dqm_sourceclient-live_cfg::path, 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_.size()) {
     //get run number from first file in the vector
     boost::filesystem::path fileName = fileNames_[0];
     std::string fileStem = fileName.stem().string();
     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 = boost::lexical_cast<long>(runStr);
         edm::LogInfo("FedRawDataInputSource") << "Autodetected run number in fileListMode -: "<< rval;
         return rval;
       }
       catch( boost::bad_lexical_cast const& ) {
         edm::LogWarning("FedRawDataInputSource") << "Unable to autodetect run number in fileListMode from file -: "<< fileName;
       }
     }
   }
   return -1;
 }

void FedRawDataInputSource::maybeOpenNewLumiSection ( const uint32_t lumiSection )

private

Definition at line 320 of file FedRawDataInputSource.cc.

References 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(), AlCaHLTBitMon_QueryRunRegistry::string, and evf::EvFDaqDirector::unlockFULocal2().

Referenced by checkNextEvent(), and getNextEvent().

 {
   if (!luminosityBlockAuxiliary()
     || luminosityBlockAuxiliary()->luminosityBlock() != lumiSection) {
 
     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);
         createBoLSFile(lumiSection,false);
         daqDirector_->unlockFULocal2();
       }
     }
     else createBoLSFile(lumiSection,true);//needed for initial lumisection
 
     currentLumiSection_ = lumiSection;
 
     resetLuminosityBlockAuxiliary();
 
     timeval tv;
     gettimeofday(&tv, 0);
     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;
   }
 }

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

inlineprivate

Definition at line 361 of file FedRawDataInputSource.cc.

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

Referenced by checkNextEvent().

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

void FedRawDataInputSource::postForkReacquireResources ( std::shared_ptr< edm::multicore::MessageReceiverForSource > )

overrideprivatevirtual

Definition at line 909 of file FedRawDataInputSource.cc.

References edm::Timestamp::beginOfTime(), edm::Timestamp::invalidTimestamp(), runNumber_, and edm::InputSource::setRunAuxiliary().

 {
   InputSource::rewind();
   setRunAuxiliary(
                   new edm::RunAuxiliary(runNumber_, edm::Timestamp::beginOfTime(),
                     edm::Timestamp::invalidTimestamp()));
 }

void FedRawDataInputSource::preForkReleaseResources ( )

overrideprivatevirtual

Reimplemented from edm::RawInputSource.

Definition at line 906 of file FedRawDataInputSource.cc.

907 {}

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

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 674 of file FedRawDataInputSource.cc.

References assert(), printConversionInfo::aux, edm::DaqProvenanceHelper::branchDescription(), checkEvery_, chunkIsFree_, InputFile::chunks_, InputFile::currentChunk_, currentFile_, currentLumiSection_, daqProvenanceHelper_, deleteFile(), edm::DaqProvenanceHelper::dummyProvenance(), eventID_, eventRunNumber_, eventsThisLumi_, fileDeleteLock_, fileListLoopMode_, filesToDelete_, fillFEDRawDataCollection(), fms_, freeChunks_, GTPEventID_, i, evf::FastMonitoringThread::inNoRequest, evf::FastMonitoringThread::inReadCleanup, evf::FastMonitoringThread::inReadEvent, L1EventID_, edm::RawInputSource::makeEvent(), evf::evtn::makeEventAuxiliary(), eostools::move(), InputFile::nProcessed_, nStreams_, edm::EventAuxiliary::PhysicsTrigger, edm::InputSource::processGUID(), processHistoryID_, edm::EventPrincipal::put(), record, evf::FastMonitoringService::setInState(), edm::EventAuxiliary::setProcessHistoryID(), streamFileTrackerPtr_, tcds_pointer_, and useL1EventID_.

Referenced by edmIntegrityCheck.PublishToFileSystem::get(), Vispa.Plugins.EdmBrowser.EdmDataAccessor.EdmDataAccessor::goto(), readNextChunkIntoBuffer(), readWorker(), and Vispa.Plugins.EdmBrowser.EdmDataAccessor.EdmDataAccessor::setFilterBranches().

 {
   if (fms_) fms_->setInState(evf::FastMonitoringThread::inReadEvent);
   std::unique_ptr<FEDRawDataCollection> rawData(new FEDRawDataCollection);
   edm::Timestamp tstamp = fillFEDRawDataCollection(*rawData);
 
   if (useL1EventID_){
     eventID_ = edm::EventID(eventRunNumber_, currentLumiSection_, L1EventID_);
     edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, true,
                 edm::EventAuxiliary::PhysicsTrigger);
     aux.setProcessHistoryID(processHistoryID_);
     makeEvent(eventPrincipal, aux);
   }
   else if(tcds_pointer_==0){
     assert(GTPEventID_);
     eventID_ = edm::EventID(eventRunNumber_, currentLumiSection_, GTPEventID_);
     edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, true,
                 edm::EventAuxiliary::PhysicsTrigger);
     aux.setProcessHistoryID(processHistoryID_);
     makeEvent(eventPrincipal, aux);
   }
   else{
     evf::evtn::TCDSRecord record((unsigned char *)(tcds_pointer_));
     edm::EventAuxiliary aux = evf::evtn::makeEventAuxiliary(&record,
                          eventRunNumber_,currentLumiSection_,
                                                  processGUID(),!fileListLoopMode_);
     aux.setProcessHistoryID(processHistoryID_);
     makeEvent(eventPrincipal, aux);
   }
 
 
 
   std::unique_ptr<edm::WrapperBase> edp(new edm::Wrapper<FEDRawDataCollection>(std::move(rawData)));
 
   //FWCore/Sources DaqProvenanceHelper before 7_1_0_pre3
   //eventPrincipal.put(daqProvenanceHelper_.constBranchDescription_, edp,
   //                   daqProvenanceHelper_.dummyProvenance_);
 
   eventPrincipal.put(daqProvenanceHelper_.branchDescription(), std::move(edp),
                      daqProvenanceHelper_.dummyProvenance());
 
   eventsThisLumi_++;
   if (fms_) fms_->setInState(evf::FastMonitoringThread::inReadCleanup);
 
   //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 == streamFileTrackerPtr_->at(i)) {
         fileIsBeingProcessed = true;
         break;
     }
       }
       if (!fileIsBeingProcessed) {
         deleteFile(it->second->fileName_);
         delete it->second;
     it = filesToDelete_.erase(it);
       }
       else it++;
     }
 
   }
   if (chunkIsFree_) freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_-1]);
   chunkIsFree_=false;
   if (fms_) fms_->setInState(evf::FastMonitoringThread::inNoRequest);
   return;
 }

void FedRawDataInputSource::readNextChunkIntoBuffer ( InputFile * file )

private

Definition at line 1338 of file FedRawDataInputSource.cc.

References bufferInputRead_, InputFile::chunkPosition_, InputFile::chunks_, eventChunkBlock_, eventChunkSize_, Exception, fileDescriptor_, InputFile::fileName_, InputFile::fileSize_, i, plotBeamSpotDB::last, LogDebug, read(), and readBlocks_.

Referenced by getNextEvent().

 {
 
   if (fileDescriptor_<0) {
     fileDescriptor_ = open(file->fileName_.c_str(), O_RDONLY);
     bufferInputRead_ = 0;
     //off_t pos = lseek(fileDescriptor,0,SEEK_SET);
     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_;
     }
   }
 
   if (file->chunkPosition_ == 0) { //in the rare case the last byte barely fit
     uint32_t existingSize = 0;
     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 {
     const uint32_t chunksize = file->chunkPosition_;
     const uint32_t blockcount=chunksize/eventChunkBlock_;
     const uint32_t leftsize = chunksize%eventChunkBlock_;
     uint32_t existingSize = eventChunkSize_ - file->chunkPosition_;
     memmove((void*) file->chunks_[0]->buf_, file->chunks_[0]->buf_ + file->chunkPosition_, existingSize);
 
     for (uint32_t i=0;i<blockcount;i++) {
       const ssize_t last = ::read(fileDescriptor_,( void*) (file->chunks_[0]->buf_ + existingSize), eventChunkBlock_);
       bufferInputRead_+=last;
       existingSize+=last;
     }
     if (leftsize) {
       const ssize_t last = ::read(fileDescriptor_,( void*)( file->chunks_[0]->buf_ + existingSize ), 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_);
       fileDescriptor_=-1;
     }
   }
 }

void FedRawDataInputSource::readSupervisor ( )

private

Definition at line 921 of file FedRawDataInputSource.cc.

References assert(), InputFile::chunks_, counter, cvReader_, cvWakeup_, daqDirector_, eventChunkSize_, fileListMode_, fileQueue_, fms_, freeChunks_, getFile(), getLSFromFilename_, grabNextJsonFile(), i, InputFile, evf::FastMonitoringThread::inRunEnd, evf::FastMonitoringThread::inSupBusy, evf::FastMonitoringThread::inSupFileLimit, evf::FastMonitoringThread::inSupLockPolling, evf::FastMonitoringThread::inSupNewFile, evf::FastMonitoringThread::inSupNewFileWaitChunk, evf::FastMonitoringThread::inSupNewFileWaitChunkCopying, evf::FastMonitoringThread::inSupNewFileWaitThread, evf::FastMonitoringThread::inSupNewFileWaitThreadCopying, evf::FastMonitoringThread::inSupNoFile, evf::FastMonitoringThread::inSupWaitFreeChunk, evf::FastMonitoringThread::inSupWaitFreeChunkCopying, evf::FastMonitoringThread::inSupWaitFreeThread, evf::FastMonitoringThread::inSupWaitFreeThreadCopying, j, LogDebug, eostools::ls(), maxBufferedFiles_, mReader_, mWakeup_, evf::EvFDaqDirector::newFile, evf::EvFDaqDirector::newLumi, evf::EvFDaqDirector::noFile, fed_dqm_sourceclient-live_cfg::path, quit_threads_, InputChunk::readComplete_, readingFilesCount_, evf::FastMonitoringService::reportLockWait(), InputChunk::reset(), evf::EvFDaqDirector::runAbort, evf::EvFDaqDirector::runEnded, setExceptionState_, evf::FastMonitoringService::setInStateSup(), edm::shutdown_flag, singleBufferMode_, evf::FastMonitoringService::startedLookingForFile(), startupCv_, startupLock_, mps_update::status, evf::FastMonitoringService::stoppedLookingForFile(), AlCaHLTBitMon_QueryRunRegistry::string, thread_quit_signal, tid_active_, mps_check::timeout, evf::EvFDaqDirector::updateFuLock(), workerJob_, workerPool_, and workerThreads_.

Referenced by checkNextEvent().

 {
   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_) fms_->setInStateSup(evf::FastMonitoringThread::inSupFileLimit);
         else if (freeChunks_.empty()) {
           if (copy_active)
             fms_->setInStateSup(evf::FastMonitoringThread::inSupWaitFreeChunkCopying);
           else
             fms_->setInStateSup(evf::FastMonitoringThread::inSupWaitFreeChunk);
         }
         else {
           if (copy_active)
             fms_->setInStateSup(evf::FastMonitoringThread::inSupWaitFreeThreadCopying);
           else
             fms_->setInStateSup(evf::FastMonitoringThread::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 (stop) break;
 
     //look for a new file
     std::string nextFile;
     uint32_t fileSize;
 
     if (fms_) {
       fms_->setInStateSup(evf::FastMonitoringThread::inSupBusy);
       fms_->startedLookingForFile();
       fms_->setInStateSup(evf::FastMonitoringThread::inSupLockPolling);
     }
 
     evf::EvFDaqDirector::FileStatus status =  evf::EvFDaqDirector::noFile;
 
     while (status == evf::EvFDaqDirector::noFile) {
       if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
     stop=true;
     break;
       }
      
       uint64_t thisLockWaitTimeUs=0.;
       if (fileListMode_) {
         //return LS if LS not set, otherwise return file
         status = getFile(ls,nextFile,fileSize,thisLockWaitTimeUs);
       }
       else
         status = daqDirector_->updateFuLock(ls,nextFile,fileSize,thisLockWaitTimeUs);
         
       if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inSupBusy);
 
       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_) {
         if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::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,fileSize,thisLockWaitTimeUs);
         if (currentLumiSection!=ls && status==evf::EvFDaqDirector::runEnded) status=evf::EvFDaqDirector::noFile;
       }
 
       if ( status == evf::EvFDaqDirector::runEnded) {
     fileQueue_.push(new InputFile(evf::EvFDaqDirector::runEnded));
     stop=true;
     break;
       }
 
       //queue new lumisection
       if( getLSFromFilename_ && ls > currentLumiSection) {
         //fms_->setInStateSup(evf::FastMonitoringThread::inSupNewLumi);
     currentLumiSection = ls;
     fileQueue_.push(new InputFile(evf::EvFDaqDirector::newLumi, currentLumiSection));
       }
 
       if( getLSFromFilename_ && currentLumiSection>0 && ls < currentLumiSection) {
           edm::LogError("FedRawDataInputSource") << "Got old LS ("<<ls<<") file from EvFDAQDirector! Expected LS:" << currentLumiSection<<". Aborting execution."<<std::endl;
       fileQueue_.push(new InputFile(evf::EvFDaqDirector::runAbort, 0));
       stop=true;
       break;
       }
 
       int dbgcount=0;
       if (status == evf::EvFDaqDirector::noFile) {
         if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inSupNoFile);
     dbgcount++;
     if (!(dbgcount%20)) LogDebug("FedRawDataInputSource") << "No file for me... sleep and try again...";
     usleep(100000);
       }
     }
     if ( status == evf::EvFDaqDirector::newFile ) {
       if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFile);
       LogDebug("FedRawDataInputSource") << "The director says to grab -: " << nextFile;
 
 
       boost::filesystem::path rawFilePath(nextFile);
       std::string 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;
     stop=true;
         break;
       }
       fileSize=st.st_size;
 
       if (fms_) {
         fms_->setInStateSup(evf::FastMonitoringThread::inSupBusy);
         fms_->stoppedLookingForFile(ls);
         fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFile);
       }
       int eventsInNewFile;
       if (fileListMode_) {
         if (fileSize==0) eventsInNewFile=0;
         else eventsInNewFile=-1;
       }
       else {
         eventsInNewFile = grabNextJsonFile(nextFile);
         assert( eventsInNewFile>=0 );
         assert((eventsInNewFile>0) == (fileSize>0));//file without events must be empty
       }
       //int eventsInNewFile = fileListMode_ ? -1 : grabNextJsonFile(nextFile);
       //if (fileListMode_ && fileSize==0) {eventsInNewFile=0;}
       //if (!fileListMode_) {
       //  assert( eventsInNewFile>=0 );
       //  assert((eventsInNewFile>0) == (fileSize>0));//file without events must be empty
       //}
 
       if (!singleBufferMode_) {
     //calculate number of needed chunks
     unsigned int neededChunks = fileSize/eventChunkSize_;
     if (fileSize%eventChunkSize_) neededChunks++;
 
         InputFile * newInputFile = new InputFile(evf::EvFDaqDirector::FileStatus::newFile,ls,rawFile,fileSize,neededChunks,eventsInNewFile,this);
         readingFilesCount_++;
         fileQueue_.push(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)
               fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFileWaitThreadCopying);
             else
               fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFileWaitThread);
           }
       //get thread
       unsigned int newTid = 0xffffffff;
       while (!workerPool_.try_pop(newTid)) {
         usleep(100000);
       }
 
           if (fms_) {
             bool copy_active=false;
             for (auto j : tid_active_) if (j) copy_active=true;
             if (copy_active)
               fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFileWaitChunkCopying);
             else
               fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFileWaitChunk);
           }
       InputChunk * newChunk = nullptr;
       while (!freeChunks_.try_pop(newChunk)) {
             usleep(100000);
             if (quit_threads_.load(std::memory_order_relaxed)) break;
       }
 
           if (newChunk == nullptr) {
             //return unused tid if we received shutdown (nullptr chunk)
             if (newTid!=0xffffffff) workerPool_.push(newTid);
             stop = true;
             break;
           }
           if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::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=newInputFile;
       workerJob_[newTid].second=newChunk;
 
       //wake up the worker thread
       cvReader_[newTid]->notify_one();
     }
       }
       else {
     if (!eventsInNewFile) {
       //still queue file for lumi update
       std::unique_lock<std::mutex> lkw(mWakeup_);
       InputFile * newInputFile = new InputFile(evf::EvFDaqDirector::FileStatus::newFile,ls,rawFile,0,0,0,this);
           readingFilesCount_++;
       fileQueue_.push(newInputFile);
       cvWakeup_.notify_one();
       return;
     }
     //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);
 
         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
         InputFile * newInputFile = new InputFile(evf::EvFDaqDirector::FileStatus::newFile,ls,rawFile,fileSize,1,eventsInNewFile,this);
         newInputFile->chunks_[0]=newChunk;
         readingFilesCount_++;
         fileQueue_.push(newInputFile);
         cvWakeup_.notify_one();
       }
     }
   }
   if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inRunEnd);
   //make sure threads finish reading
   unsigned numFinishedThreads = 0;
   while (numFinishedThreads < workerThreads_.size()) {
     unsigned int tid;
     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];
   }
 }

void FedRawDataInputSource::readWorker ( unsigned int tid )

private

Definition at line 1203 of file FedRawDataInputSource.cc.

References assert(), InputChunk::buf_, InputFile::chunks_, cvReader_, detectedFRDversion_, mps_update::diff, end, eventChunkBlock_, mergeVDriftHistosByStation::file, InputChunk::fileIndex_, InputFile::fileName_, plotBeamSpotDB::first, i, init, plotBeamSpotDB::last, LogDebug, mReader_, fileCollector::now, InputChunk::offset_, read(), readBlocks_, InputChunk::readComplete_, edm::second(), setExceptionState_, dqm_diff::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 (1) {
 
     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;
 
     int fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
     off_t pos = lseek(fileDescriptor,chunk->offset_,SEEK_SET);
 
 
     if (fileDescriptor>=0)
       LogDebug("FedRawDataInputSource") << "Reader thread opened file -: TID: " << tid << " file: " << file->fileName_ << " at offset " << pos;
     else
     {
       edm::LogError("FedRawDataInputSource") <<
       "readWorker failed to open file -: " << file->fileName_ << " fd:" << fileDescriptor <<
       " or seek to offset " << chunk->offset_ << ", lseek returned:" << pos;
       setExceptionState_=true;
       return;
 
     }
 
     unsigned int bufferLeft = 0;
     auto start = std::chrono::high_resolution_clock::now();
     for (unsigned int i=0;i<readBlocks_;i++)
     {
       const ssize_t last = ::read(fileDescriptor,( void*) (chunk->buf_+bufferLeft), eventChunkBlock_);
       if ( last > 0 )
     bufferLeft+=last;
       if (last < eventChunkBlock_) {
     assert(chunk->usedSize_==i*eventChunkBlock_+last);
     break;
       }
     }
     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)";
     close(fileDescriptor);
 
     if (detectedFRDversion_==0 && chunk->offset_==0) detectedFRDversion_=*((uint32*)chunk->buf_);
     assert(detectedFRDversion_<=5);
     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
 
   }
 }

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

private

Definition at line 1392 of file FedRawDataInputSource.cc.

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

Referenced by checkNextEvent(), 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;
 }

void FedRawDataInputSource::rewind_ ( )

overrideprivatevirtual

Reimplemented from edm::RawInputSource.

Definition at line 917 of file FedRawDataInputSource.cc.

918 {}

void FedRawDataInputSource::threadError ( )

private

Definition at line 1278 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

friend class InputChunk

friend

Definition at line 46 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

friend class InputFile

friend

Open Root file and provide MEs ############.

Definition at line 45 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

Member Data Documentation

uint32_t FedRawDataInputSource::bufferInputRead_ = 0

private

Definition at line 179 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readNextChunkIntoBuffer().

unsigned int FedRawDataInputSource::checkEvery_ = 10

private

Definition at line 170 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

bool FedRawDataInputSource::chunkIsFree_ =false

private

Definition at line 142 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

InputFile* FedRawDataInputSource::currentFile_ = 0

private

Definition at line 141 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

int FedRawDataInputSource::currentFileIndex_ = -1

private

Definition at line 164 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

unsigned int FedRawDataInputSource::currentLumiSection_

private

Definition at line 122 of file FedRawDataInputSource.h.

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

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

private

Definition at line 155 of file FedRawDataInputSource.h.

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

std::condition_variable FedRawDataInputSource::cvWakeup_

private

Definition at line 174 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

evf::EvFDaqDirector* FedRawDataInputSource::daqDirector_ = 0

private

Definition at line 87 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), createBoLSFile(), FedRawDataInputSource(), getNextEvent(), grabNextJsonFile(), maybeOpenNewLumiSection(), and readSupervisor().

const edm::DaqProvenanceHelper FedRawDataInputSource::daqProvenanceHelper_

private

Definition at line 115 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and read().

std::string FedRawDataInputSource::defPath_

private

Definition at line 89 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

uint32 FedRawDataInputSource::detectedFRDversion_ =0

private

Definition at line 140 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readWorker().

jsoncollector::DataPointDefinition* FedRawDataInputSource::dpd_

private

Definition at line 130 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and grabNextJsonFile().

std::unique_ptr<FRDEventMsgView> FedRawDataInputSource::event_

private

Definition at line 117 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), fillFEDRawDataCollection(), and getNextEvent().

unsigned int FedRawDataInputSource::eventChunkBlock_

private

Definition at line 92 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::eventChunkSize_

private

Definition at line 91 of file FedRawDataInputSource.h.

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

edm::EventID FedRawDataInputSource::eventID_

private

Definition at line 119 of file FedRawDataInputSource.h.

Referenced by read().

uint32_t FedRawDataInputSource::eventRunNumber_ =0

private

Definition at line 123 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and read().

unsigned int FedRawDataInputSource::eventsThisLumi_

private

Definition at line 127 of file FedRawDataInputSource.h.

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

unsigned long FedRawDataInputSource::eventsThisRun_ = 0

private

Definition at line 128 of file FedRawDataInputSource.h.

std::mutex FedRawDataInputSource::fileDeleteLock_

private

Definition at line 167 of file FedRawDataInputSource.h.

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

int FedRawDataInputSource::fileDescriptor_ = -1

private

Definition at line 178 of file FedRawDataInputSource.h.

Referenced by readNextChunkIntoBuffer().

unsigned int FedRawDataInputSource::fileListIndex_ = 0

private

Definition at line 108 of file FedRawDataInputSource.h.

Referenced by getFile().

const bool FedRawDataInputSource::fileListLoopMode_

private

Definition at line 109 of file FedRawDataInputSource.h.

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

const bool FedRawDataInputSource::fileListMode_

private

Definition at line 107 of file FedRawDataInputSource.h.

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

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

private

Definition at line 104 of file FedRawDataInputSource.h.

Referenced by getFile(), and initFileList().

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

private

Definition at line 166 of file FedRawDataInputSource.h.

tbb::concurrent_queue<InputFile*> FedRawDataInputSource::fileQueue_

private

Definition at line 152 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

std::list<std::pair<int,InputFile*> > FedRawDataInputSource::filesToDelete_

private

Definition at line 165 of file FedRawDataInputSource.h.

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

evf::FastMonitoringService* FedRawDataInputSource::fms_ = 0

private

Definition at line 86 of file FedRawDataInputSource.h.

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

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

private

Definition at line 151 of file FedRawDataInputSource.h.

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

std::string FedRawDataInputSource::fuOutputDir_

private

Definition at line 113 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and grabNextJsonFile().

const bool FedRawDataInputSource::getLSFromFilename_

private

Definition at line 100 of file FedRawDataInputSource.h.

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

uint32_t FedRawDataInputSource::GTPEventID_ = 0

private

Definition at line 124 of file FedRawDataInputSource.h.

Referenced by fillFEDRawDataCollection(), and read().

uint32_t FedRawDataInputSource::L1EventID_ = 0

private

Definition at line 125 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and read().

unsigned int FedRawDataInputSource::loopModeIterationInc_ = 0

private

Definition at line 110 of file FedRawDataInputSource.h.

Referenced by getFile().

unsigned int FedRawDataInputSource::maxBufferedFiles_

private

Definition at line 95 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

std::mutex FedRawDataInputSource::monlock_

private

Definition at line 184 of file FedRawDataInputSource.h.

Referenced by getEventReport(), and reportEventsThisLumiInSource().

std::mutex FedRawDataInputSource::mReader_

private

Definition at line 154 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::mWakeup_

private

Definition at line 173 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

unsigned int FedRawDataInputSource::nStreams_ = 0

private

Definition at line 169 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

unsigned int FedRawDataInputSource::numBuffers_

private

Definition at line 94 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

unsigned int FedRawDataInputSource::numConcurrentReads_

private

Definition at line 96 of file FedRawDataInputSource.h.

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

edm::ProcessHistoryID FedRawDataInputSource::processHistoryID_

private

Definition at line 120 of file FedRawDataInputSource.h.

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

std::atomic<bool> FedRawDataInputSource::quit_threads_

private

Definition at line 158 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::readBlocks_

private

Definition at line 93 of file FedRawDataInputSource.h.

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

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

private

Definition at line 97 of file FedRawDataInputSource.h.

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

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

private

Definition at line 145 of file FedRawDataInputSource.h.

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

edm::RunNumber_t FedRawDataInputSource::runNumber_

private

Definition at line 112 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), FedRawDataInputSource(), and postForkReacquireResources().

bool FedRawDataInputSource::setExceptionState_ = false

private

Definition at line 160 of file FedRawDataInputSource.h.

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

bool FedRawDataInputSource::singleBufferMode_

private

Definition at line 177 of file FedRawDataInputSource.h.

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

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

private

Definition at line 183 of file FedRawDataInputSource.h.

Referenced by getEventReport(), and reportEventsThisLumiInSource().

bool FedRawDataInputSource::startedSupervisorThread_ = false

private

Definition at line 144 of file FedRawDataInputSource.h.

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

std::condition_variable FedRawDataInputSource::startupCv_

private

Definition at line 162 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::startupLock_

private

Definition at line 161 of file FedRawDataInputSource.h.

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

std::vector<int>* FedRawDataInputSource::streamFileTrackerPtr_ = 0

private

Definition at line 168 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

unsigned char* FedRawDataInputSource::tcds_pointer_

private

Definition at line 126 of file FedRawDataInputSource.h.

Referenced by fillFEDRawDataCollection(), and read().

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

private

Definition at line 159 of file FedRawDataInputSource.h.

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

std::atomic<bool> FedRawDataInputSource::threadInit_

private

Definition at line 181 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and readWorker().

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

private

Definition at line 156 of file FedRawDataInputSource.h.

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

const bool FedRawDataInputSource::useL1EventID_

private

Definition at line 103 of file FedRawDataInputSource.h.

Referenced by read().

const bool FedRawDataInputSource::verifyAdler32_

private

Definition at line 101 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

const bool FedRawDataInputSource::verifyChecksum_

private

Definition at line 102 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

std::vector<ReaderInfo> FedRawDataInputSource::workerJob_

private

Definition at line 149 of file FedRawDataInputSource.h.

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

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

private

Definition at line 148 of file FedRawDataInputSource.h.

Referenced by readSupervisor(), and readWorker().

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

Constructor & Destructor Documentation

Member Function Documentation

Friends And Related Function Documentation

Member Data Documentation