FedRawDataInputSource Class Reference

#include <FedRawDataInputSource.h>

Inheritance diagram for FedRawDataInputSource:

Public Member Functions
	FedRawDataInputSource (edm::ParameterSet const &, edm::InputSourceDescription const &)
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 >	branchIDListHelper () const
	Accessor for branchIDListHelper. More...
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
bool	primary () const
	Accessor for primary input source flag. More...
ProcessConfiguration const &	processConfiguration () const
	Accessor for Process Configuration. More...
std::string const &	processGUID () const
	Accessor for global process identifier. More...
ProcessHistoryRegistry const &	processHistoryRegistry () const
	Const accessor for process history registry. More...
ProcessHistoryRegistry &	processHistoryRegistryForUpdate ()
	Non-const accessor for process history registry. More...
ProcessingMode	processingMode () const
	RunsLumisAndEvents (default), RunsAndLumis, or Runs. More...
std::shared_ptr < ProductRegistry const >	productRegistry () const
	Accessor 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::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 () const
	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 >	thinnedAssociationsHelper () const
	Accessor for thinnedAssociationsHelper. More...
Timestamp const &	timestamp () const
	Accessor for the current time, as seen by the input source. More...
virtual	~InputSource ()
	Destructor. More...

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
std::shared_ptr < LuminosityBlockPrincipal > const	luminosityBlockPrincipal () const
bool	newLumi () const
bool	newRun () const
ProcessHistoryRegistry &	processHistoryRegistryUpdate () const
ProductRegistry &	productRegistryUpdate () const
void	reset () const
void	resetEventCached ()
void	resetLuminosityBlockAuxiliary (bool isNewLumi=true) const
void	resetNewLumi ()
void	resetNewRun ()
void	resetRunAuxiliary (bool isNewRun=true) const
std::shared_ptr< RunPrincipal > const	runPrincipal () 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	getNextEvent ()
int	grabNextJsonFile (boost::filesystem::path const &)
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)
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
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_
const std::string	fuOutputDir_
const bool	getLSFromFilename_
uint32_t	GTPEventID_ = 0
uint32_t	L1EventID_ = 0
unsigned int	maxBufferedFiles_
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_
const edm::RunNumber_t	runNumber_
bool	setExceptionState_ = false
bool	singleBufferMode_
bool	startedSupervisorThread_ = false
std::condition_variable	startupCv_
std::mutex	startupLock_
std::vector< int > *	streamFileTrackerPtr_ = 0
unsigned char *	tcds_pointer_
std::vector< bool >	thread_quit_signal
std::atomic< bool >	threadInit_
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
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 &)

Detailed Description

Definition at line 43 of file FedRawDataInputSource.h.

Member Typedef Documentation

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

private

Definition at line 124 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_, edm::hlt::Exception, fileDeleteLock_, filesToDelete_, fms_, freeChunks_, i, InputChunk, edm::Timestamp::invalidTimestamp(), numBuffers_, numConcurrentReads_, processHistoryID_, edm::InputSource::processHistoryRegistryForUpdate(), edm::InputSource::productRegistryUpdate(), quit_threads_, readBlocks_, readingFilesCount_, readWorker(), edm::InputSource::runAuxiliary(), runNumber_, evf::EvFDaqDirector::setDeleteTracking(), evf::EvFDaqDirector::setFMS(), edm::InputSource::setNewRun(), edm::InputSource::setRunAuxiliary(), singleBufferMode_, startupCv_, startupLock_, AlCaHLTBitMon_QueryRunRegistry::string, thread_quit_signal, threadInit_, 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",16)*1048576),
  eventChunkBlock_(pset.getUntrackedParameter<unsigned int> ("eventChunkBlock",eventChunkSize_/1048576)*1048576),
  numBuffers_(pset.getUntrackedParameter<unsigned int> ("numBuffers",1)),
  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)),
  runNumber_(edm::Service<evf::EvFDaqDirector>()->getRunNumber()),
  fuOutputDir_(edm::Service<evf::EvFDaqDirector>()->baseRunDir()),
  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;

  processHistoryID_ = daqProvenanceHelper_.daqInit(productRegistryUpdate(), processHistoryRegistryForUpdate());
  setNewRun();
  setRunAuxiliary(new edm::RunAuxiliary(runNumber_, edm::Timestamp::beginOfTime(),
                    edm::Timestamp::invalidTimestamp()));

  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";

  //het handles to DaqDirector and FastMonitoringService because it isn't acessible in readSupervisor thread

  try {
    fms_ = (evf::FastMonitoringService *) (edm::Service<evf::MicroStateService>().operator->());
  } catch (...){
    edm::LogWarning("FedRawDataInputSource") << "FastMonitoringService not found";
    assert(0);//test
  }

  try {
    daqDirector_ = (evf::EvFDaqDirector *) (edm::Service<evf::EvFDaqDirector>().operator->());
    //set DaqDirector to delete files in preGlobalEndLumi callback
    daqDirector_->setDeleteTracking(&fileDeleteLock_,&filesToDelete_);
    if (fms_) daqDirector_->setFMS(fms_);
  } catch (...){
    edm::LogWarning("FedRawDataInputSource") << "EvFDaqDirector not found";
    assert(0);//test
  }

  //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);
    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 149 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 182 of file FedRawDataInputSource.cc.

References evf::EvFDaqDirector::createProcessingNotificationMaybe(), currentLumiSection_, daqDirector_, evf::EvFDaqDirector::emptyLumisectionMode(), event_, eventRunNumber_, eventsThisLumi_, fms_, newFWLiteAna::found, evf::EvFDaqDirector::getEoLSFilePathOnBU(), evf::EvFDaqDirector::getEoLSFilePathOnFU(), evf::EvFDaqDirector::getEoRFilePathOnFU(), getLSFromFilename_, L1EventID_, evf::EvFDaqDirector::lockFULocal2(), maybeOpenNewLumiSection(), evf::EvFDaqDirector::newLumi, nextEvent(), evf::EvFDaqDirector::noFile, readSupervisor(), readSupervisorThread_, evf::FastMonitoringService::reportEventsThisLumiInSource(), edm::InputSource::resetLuminosityBlockAuxiliary(), evf::EvFDaqDirector::runEnded, edm::InputSource::setEventCached(), startedSupervisorThread_, startupCv_, startupLock_, AlCaHLTBitMon_QueryRunRegistry::string, and evf::EvFDaqDirector::unlockFULocal2().

{
  if (!startedSupervisorThread_)
  {
    //this thread opens new files and dispatches reading to worker readers
    //threadInit_.store(false,std::memory_order_release);
    std::unique_lock<std::mutex> lk(startupLock_);
    readSupervisorThread_.reset(new std::thread(&FedRawDataInputSource::readSupervisor,this));
    startedSupervisorThread_=true;
    startupCv_.wait(lk);
  }
  //signal hltd to start event accounting
  if (!currentLumiSection_ && daqDirector_->emptyLumisectionMode())
    daqDirector_->createProcessingNotificationMaybe();

  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);
      }
      if (fms_) fms_->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_) {
          if (fms_) fms_->reportEventsThisLumiInSource(currentLumiSection_,eventsThisLumi_);
      eventsThisLumi_=0;
          maybeOpenNewLumiSection( event_->lumi() );
    }
      }
      eventRunNumber_=event_->run();
      L1EventID_ = event_->event();

      setEventCached();

      return true;
    }
  }
}

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

private

Definition at line 254 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 577 of file FedRawDataInputSource.cc.

References cppFunctionSkipper::exception, convertXMLtoSQLite_cfg::fileName, LogDebug, cmsHarvester::path, and python.multivaluedict::remove().

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

{
  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 (...) {/*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 (...) {/*file gets deleted first time but exception is still thrown*/}
  }
}

bool FedRawDataInputSource::exceptionState ( )

inlineprivate

Definition at line 72 of file FedRawDataInputSource.h.

References setExceptionState_.

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

{return setExceptionState_;}

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

private

Definition at line 676 of file FedRawDataInputSource.cc.

References assert(), FEDRawData::data(), event(), event_, fedt_struct::eventsize, evf::evtn::evm_board_sense(), edm::hlt::Exception, FED_EVSZ_EXTRACT, FED_SOID_EXTRACT, FEDRawDataCollection::FEDData(), HLT_25ns14e33_v1_cff::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, tcds_pointer_, and cond::rpcobgas::time.

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

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

inlineprivate

Definition at line 316 of file FedRawDataInputSource.cc.

References InputFile::advance(), assert(), bufferInputRead_, InputFile::bufferPosition_, checkEvery_, chunkIsFree_, InputFile::chunkPosition_, InputFile::chunks_, crc32c(), InputFile::currentChunk_, currentFile_, currentFileIndex_, currentLumiSection_, cvWakeup_, daqDirector_, deleteFile(), detectedFRDversion_, event_, eventChunkSize_, eventsThisLumi_, edm::hlt::Exception, exceptionState(), fileDeleteLock_, InputFile::fileName_, fileQueue_, InputFile::fileSize_, filesToDelete_, fms_, FRDHeaderVersionSize, freeChunks_, getLSFromFilename_, evf::EvFDaqDirector::getStreamFileTracker(), evf::EvFDaqDirector::isSingleStreamThread(), dttmaxenums::L, InputFile::lumi_, maybeOpenNewLumiSection(), InputFile::moveToPreviousChunk(), mWakeup_, InputFile::nChunks_, InputFile::nEvents_, evf::EvFDaqDirector::newFile, evf::EvFDaqDirector::newLumi, evf::EvFDaqDirector::noFile, InputFile::nProcessed_, nStreams_, InputFile::parent_, readingFilesCount_, readNextChunkIntoBuffer(), evf::FastMonitoringService::reportEventsThisLumiInSource(), InputFile::rewindChunk(), evf::EvFDaqDirector::runAbort, evf::EvFDaqDirector::runEnded, evf::EvFDaqDirector::sameFile, evf::FastMonitoringService::setExceptionDetected(), setExceptionState_, singleBufferMode_, ntuplemaker::status, InputFile::status_, streamFileTrackerPtr_, threadError(), evf::EvFDaqDirector::updateFileIndex(), verifyAdler32_, verifyChecksum_, and InputFile::waitForChunk().

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 (!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)
    {
      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 (getLSFromFilename_) {
    if (currentFile_->lumi_ > currentLumiSection_) {
          if (fms_) fms_->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);
  }

  //file is empty
  if (!currentFile_->fileSize_) {
    readingFilesCount_--;
    //try to open new lumi
    assert(currentFile_->nChunks_==0);
    if (getLSFromFilename_)
      if (currentFile_->lumi_ > currentLumiSection_) {
        if (fms_) fms_->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_!=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()) {
      //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
    while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
      usleep(10000);
      if (currentFile_->parent_->exceptionState()) currentFile_->parent_->threadError();
    }

    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
    while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
      usleep(10000);
      if (setExceptionState_) threadError();
    }

    //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 (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;
    }
  }


  currentFile_->nProcessed_++;

  return evf::EvFDaqDirector::sameFile;
}

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

private

Definition at line 730 of file FedRawDataInputSource.cc.

References filterCSVwithJSON::copy, daqDirector_, data, jsoncollector::DataPoint::deserialize(), dpd_, alignCSCRings::e, cppFunctionSkipper::exception, edm::hlt::Exception, fuOutputDir_, jsoncollector::DataPoint::getData(), jsoncollector::DataPointDefinition::getNames(), i, LogDebug, Json::Reader::parse(), cmsHarvester::path, matplotRender::reader, python.multivaluedict::remove(), 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;

    if (!reader.parse(ij, deserializeRoot))
      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;
}

void FedRawDataInputSource::maybeOpenNewLumiSection ( const uint32_t  lumiSection )

private

Definition at line 265 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 306 of file FedRawDataInputSource.cc.

References getNextEvent(), svgfig::load(), evf::EvFDaqDirector::noFile, edm::shutdown_flag, and ntuplemaker::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 833 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 830 of file FedRawDataInputSource.cc.

{}

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

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 607 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_, filesToDelete_, fillFEDRawDataCollection(), freeChunks_, GTPEventID_, i, L1EventID_, edm::RawInputSource::makeEvent(), evf::evtn::makeEventAuxiliary(), eostools::move(), InputFile::nProcessed_, nStreams_, edm::EventAuxiliary::PhysicsTrigger, edm::InputSource::processGUID(), processHistoryID_, edm::EventPrincipal::put(), record, edm::EventAuxiliary::setProcessHistoryID(), streamFileTrackerPtr_, tcds_pointer_, and useL1EventID_.

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

{
  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());
    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_++;

  //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;
  return;
}

void FedRawDataInputSource::readNextChunkIntoBuffer ( InputFile *  file )

private

Definition at line 1183 of file FedRawDataInputSource.cc.

References bufferInputRead_, InputFile::chunkPosition_, InputFile::chunks_, eventChunkBlock_, eventChunkSize_, edm::hlt::Exception, fileDescriptor_, InputFile::fileName_, InputFile::fileSize_, i, prof2calltree::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;
      existingSize+=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 845 of file FedRawDataInputSource.cc.

References assert(), InputFile::chunks_, counter, cvReader_, cvWakeup_, daqDirector_, eventChunkSize_, fileQueue_, fms_, freeChunks_, getLSFromFilename_, grabNextJsonFile(), i, InputFile, LogDebug, eostools::ls(), maxBufferedFiles_, mReader_, mWakeup_, evf::EvFDaqDirector::newFile, evf::EvFDaqDirector::newLumi, evf::EvFDaqDirector::noFile, cmsHarvester::path, quit_threads_, InputChunk::readComplete_, readingFilesCount_, evf::FastMonitoringService::reportLockWait(), InputChunk::reset(), evf::EvFDaqDirector::runAbort, evf::EvFDaqDirector::runEnded, edm::shutdown_flag, singleBufferMode_, evf::FastMonitoringService::startedLookingForFile(), startupCv_, startupLock_, ntuplemaker::status, evf::FastMonitoringService::stoppedLookingForFile(), AlCaHLTBitMon_QueryRunRegistry::string, thread_quit_signal, 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();
  }

  while (!stop) {

    //wait for at least one free thread and chunk
    int counter=0;
    while ((workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty() || readingFilesCount_>=maxBufferedFiles_)
    {
      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 ls;
    uint32_t fileSize;

    uint32_t monLS=1;
    uint32_t lockCount=0;
    uint64_t sumLockWaitTimeUs=0.;

    if (fms_) fms_->startedLookingForFile();

    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.;
      status = daqDirector_->updateFuLock(ls,nextFile,fileSize,thisLockWaitTimeUs);

      //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) {
        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 ( status == evf::EvFDaqDirector::runEnded) {
    fileQueue_.push(new InputFile(evf::EvFDaqDirector::runEnded));
    stop=true;
    break;
      }

      //queue new lumisection
      if( getLSFromFilename_ && ls > currentLumiSection) {
    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) {
    dbgcount++;
    if (!(dbgcount%20)) LogDebug("FedRawDataInputSource") << "No file for me... sleep and try again...";
    usleep(100000);
      }
    }
    if ( status == evf::EvFDaqDirector::newFile ) {
      LogDebug("FedRawDataInputSource") << "The director says to grab -: " << nextFile;


      boost::filesystem::path rawFilePath(nextFile);
      std::string rawFile = rawFilePath.replace_extension(".raw").string();

      struct stat st;
      stat(rawFile.c_str(),&st);
      fileSize=st.st_size;

      int eventsInNewFile = grabNextJsonFile(nextFile);
      if (fms_) fms_->stoppedLookingForFile(ls);
      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++) {

      //get thread
      unsigned int newTid = 0xffffffff;
      while (!workerPool_.try_pop(newTid)) {
        usleep(100000);
      }

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

      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();
      }
    }
  }
  //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 1050 of file FedRawDataInputSource.cc.

References assert(), InputChunk::buf_, InputFile::chunks_, cvReader_, detectedFRDversion_, diffTreeTool::diff, end, eventChunkBlock_, mergeVDriftHistosByStation::file, InputChunk::fileIndex_, InputFile::fileName_, plotBeamSpotDB::first, i, init, prof2calltree::last, LogDebug, mReader_, fileCollector::now, InputChunk::offset_, read(), readBlocks_, InputChunk::readComplete_, edm::second(), setExceptionState_, dqm_diff::start, startupCv_, startupLock_, thread_quit_signal, threadInit_, InputChunk::usedSize_, workerJob_, and workerPool_.

Referenced by FedRawDataInputSource().

{
  bool init = true;
  threadInit_.exchange(true,std::memory_order_acquire);

  while (1) {

    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;

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

overrideprivatevirtual

Reimplemented from edm::RawInputSource.

Definition at line 841 of file FedRawDataInputSource.cc.

{}

void FedRawDataInputSource::threadError ( )

private

Definition at line 1123 of file FedRawDataInputSource.cc.

References edm::hlt::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 164 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readNextChunkIntoBuffer().

unsigned int FedRawDataInputSource::checkEvery_ = 10

private

Definition at line 155 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

bool FedRawDataInputSource::chunkIsFree_ =false

private

Definition at line 128 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

InputFile* FedRawDataInputSource::currentFile_ = 0

private

Definition at line 127 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

int FedRawDataInputSource::currentFileIndex_ = -1

private

Definition at line 149 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

unsigned int FedRawDataInputSource::currentLumiSection_

private

Definition at line 108 of file FedRawDataInputSource.h.

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

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

private

Definition at line 141 of file FedRawDataInputSource.h.

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

std::condition_variable FedRawDataInputSource::cvWakeup_

private

Definition at line 159 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

evf::EvFDaqDirector* FedRawDataInputSource::daqDirector_ = 0

private

Definition at line 79 of file FedRawDataInputSource.h.

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

const edm::DaqProvenanceHelper FedRawDataInputSource::daqProvenanceHelper_

private

Definition at line 101 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and read().

std::string FedRawDataInputSource::defPath_

private

Definition at line 81 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

uint32 FedRawDataInputSource::detectedFRDversion_ =0

private

Definition at line 126 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readWorker().

jsoncollector::DataPointDefinition* FedRawDataInputSource::dpd_

private

Definition at line 116 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and grabNextJsonFile().

std::unique_ptr<FRDEventMsgView> FedRawDataInputSource::event_

private

Definition at line 103 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::eventChunkBlock_

private

Definition at line 84 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::eventChunkSize_

private

Definition at line 83 of file FedRawDataInputSource.h.

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

edm::EventID FedRawDataInputSource::eventID_

private

Definition at line 105 of file FedRawDataInputSource.h.

Referenced by read().

uint32_t FedRawDataInputSource::eventRunNumber_ =0

private

Definition at line 109 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and read().

unsigned int FedRawDataInputSource::eventsThisLumi_

private

Definition at line 113 of file FedRawDataInputSource.h.

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

unsigned long FedRawDataInputSource::eventsThisRun_ = 0

private

Definition at line 114 of file FedRawDataInputSource.h.

std::mutex FedRawDataInputSource::fileDeleteLock_

private

Definition at line 152 of file FedRawDataInputSource.h.

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

int FedRawDataInputSource::fileDescriptor_ = -1

private

Definition at line 163 of file FedRawDataInputSource.h.

Referenced by readNextChunkIntoBuffer().

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

private

Definition at line 151 of file FedRawDataInputSource.h.

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

private

Definition at line 138 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

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

private

Definition at line 150 of file FedRawDataInputSource.h.

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

evf::FastMonitoringService* FedRawDataInputSource::fms_ = 0

private

Definition at line 78 of file FedRawDataInputSource.h.

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

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

private

Definition at line 137 of file FedRawDataInputSource.h.

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

const std::string FedRawDataInputSource::fuOutputDir_

private

Definition at line 99 of file FedRawDataInputSource.h.

Referenced by grabNextJsonFile().

const bool FedRawDataInputSource::getLSFromFilename_

private

Definition at line 92 of file FedRawDataInputSource.h.

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

uint32_t FedRawDataInputSource::GTPEventID_ = 0

private

Definition at line 110 of file FedRawDataInputSource.h.

Referenced by fillFEDRawDataCollection(), and read().

uint32_t FedRawDataInputSource::L1EventID_ = 0

private

Definition at line 111 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and read().

unsigned int FedRawDataInputSource::maxBufferedFiles_

private

Definition at line 87 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

std::mutex FedRawDataInputSource::mReader_

private

Definition at line 140 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::mWakeup_

private

Definition at line 158 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

unsigned int FedRawDataInputSource::nStreams_ = 0

private

Definition at line 154 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

unsigned int FedRawDataInputSource::numBuffers_

private

Definition at line 86 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

unsigned int FedRawDataInputSource::numConcurrentReads_

private

Definition at line 88 of file FedRawDataInputSource.h.

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

edm::ProcessHistoryID FedRawDataInputSource::processHistoryID_

private

Definition at line 106 of file FedRawDataInputSource.h.

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

std::atomic<bool> FedRawDataInputSource::quit_threads_

private

Definition at line 143 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::readBlocks_

private

Definition at line 85 of file FedRawDataInputSource.h.

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

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

private

Definition at line 89 of file FedRawDataInputSource.h.

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

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

private

Definition at line 131 of file FedRawDataInputSource.h.

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

const edm::RunNumber_t FedRawDataInputSource::runNumber_

private

Definition at line 97 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and postForkReacquireResources().

bool FedRawDataInputSource::setExceptionState_ = false

private

Definition at line 145 of file FedRawDataInputSource.h.

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

bool FedRawDataInputSource::singleBufferMode_

private

Definition at line 162 of file FedRawDataInputSource.h.

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

bool FedRawDataInputSource::startedSupervisorThread_ = false

private

Definition at line 130 of file FedRawDataInputSource.h.

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

std::condition_variable FedRawDataInputSource::startupCv_

private

Definition at line 147 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::startupLock_

private

Definition at line 146 of file FedRawDataInputSource.h.

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

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

private

Definition at line 153 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

unsigned char* FedRawDataInputSource::tcds_pointer_

private

Definition at line 112 of file FedRawDataInputSource.h.

Referenced by fillFEDRawDataCollection(), and read().

std::vector<bool> FedRawDataInputSource::thread_quit_signal

private

Definition at line 144 of file FedRawDataInputSource.h.

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

std::atomic<bool> FedRawDataInputSource::threadInit_

private

Definition at line 166 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and readWorker().

const bool FedRawDataInputSource::useL1EventID_

private

Definition at line 95 of file FedRawDataInputSource.h.

Referenced by read().

const bool FedRawDataInputSource::verifyAdler32_

private

Definition at line 93 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

const bool FedRawDataInputSource::verifyChecksum_

private

Definition at line 94 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

std::vector<ReaderInfo> FedRawDataInputSource::workerJob_

private

Definition at line 135 of file FedRawDataInputSource.h.

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

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

private

Definition at line 134 of file FedRawDataInputSource.h.

Referenced by readSupervisor(), and readWorker().

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

private

Definition at line 132 of file FedRawDataInputSource.h.

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