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
boost::shared_ptr < ActivityRegistry >	actReg () const
	Accessor for Activity Registry. More...
boost::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 (boost::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...
boost::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...
boost::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...
boost::shared_ptr < LuminosityBlockAuxiliary >	readLuminosityBlockAuxiliary ()
	Read next luminosity block Auxilary. More...
void	readRun (RunPrincipal &runPrincipal, HistoryAppender &historyAppender)
	Read next run (new run) More...
boost::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...
boost::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 ()
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
boost::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
boost::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	deleteFile (std::string const &)
bool	exceptionState ()
edm::Timestamp	fillFEDRawDataCollection (std::auto_ptr< FEDRawDataCollection > &) const
evf::EvFDaqDirector::FileStatus	getNextEvent ()
int	grabNextJsonFile (boost::filesystem::path const &)
void	maybeOpenNewLumiSection (const uint32_t lumiSection)
evf::EvFDaqDirector::FileStatus	nextEvent ()
virtual void	postForkReacquireResources (boost::shared_ptr< edm::multicore::MessageReceiverForSource >) override
virtual void	preForkReleaseResources () override
void	readNextChunkIntoBuffer (InputFile *file)
void	readSupervisor ()
void	readWorker (unsigned int tid)
void	renameToNextFree (std::string const &fileName) const
virtual void	rewind_ () override
void	threadError ()

Private Attributes
uint32_t	bufferInputRead_ = 0
unsigned int	checkEvery_ = 10
bool	chunkIsFree_ =false
InputFile *	currentFile_ = nullptr
int	currentFileIndex_ = -1
unsigned int	currentLumiSection_
std::vector < std::condition_variable * >	cvReader_
std::condition_variable	cvWakeup_
evf::EvFDaqDirector *	daqDirector_ =nullptr
const edm::DaqProvenanceHelper	daqProvenanceHelper_
std::string	defPath_
jsoncollector::DataPointDefinition *	dpd_
std::unique_ptr< FRDEventMsgView >	event_
unsigned int	eventChunkBlock_
unsigned int	eventChunkSize_
edm::EventID	eventID_
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_ =nullptr
tbb::concurrent_queue < InputChunk * >	freeChunks_
const std::string	fuOutputDir_
const bool	getLSFromFilename_
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::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_ = nullptr
const bool	testModeNoBuilderUnit_
std::vector< bool >	thread_quit_signal
std::atomic< bool >	threadInit_
const bool	verifyAdler32_
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 39 of file FedRawDataInputSource.h.

Member Typedef Documentation

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

private

Definition at line 113 of file FedRawDataInputSource.h.

Constructor & Destructor Documentation

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

explicit

Definition at line 49 of file FedRawDataInputSource.cc.

References edm::Timestamp::beginOfTime(), 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_, readWorker(), edm::InputSource::runAuxiliary(), runNumber_, evf::EvFDaqDirector::setDeleteTracking(), evf::EvFDaqDirector::setFMS(), edm::InputSource::setNewRun(), edm::InputSource::setRunAuxiliary(), singleBufferMode_, startupCv_, startupLock_, AlCaHLTBitMon_QueryRunRegistry::string, testModeNoBuilderUnit_, 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)),
  getLSFromFilename_(pset.getUntrackedParameter<bool> ("getLSFromFilename", true)),
  verifyAdler32_(pset.getUntrackedParameter<bool> ("verifyAdler32", true)),
  testModeNoBuilderUnit_(edm::Service<evf::EvFDaqDirector>()->getTestModeNoBuilderUnit()),
  runNumber_(edm::Service<evf::EvFDaqDirector>()->getRunNumber()),
  fuOutputDir_(edm::Service<evf::EvFDaqDirector>()->baseRunDir()),
  daqProvenanceHelper_(edm::TypeID(typeid(FEDRawDataCollection))),
  eventID_(),
  processHistoryID_(),
  currentLumiSection_(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;
  if (testModeNoBuilderUnit_)
    edm::LogInfo("FedRawDataInputSource") << "Test mode is ON!";

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

  //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
    if (!testModeNoBuilderUnit_)
      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 140 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 173 of file FedRawDataInputSource.cc.

References currentLumiSection_, daqDirector_, event_, eventID_, eventsThisLumi_, fms_, newFWLiteAna::found, evf::EvFDaqDirector::getEoLSFilePathOnBU(), evf::EvFDaqDirector::getEoLSFilePathOnFU(), evf::EvFDaqDirector::getEoRFilePathOnFU(), getLSFromFilename_, 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_, and AlCaHLTBitMon_QueryRunRegistry::string.

{
  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);
  }
  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_->lockFULocal2();
          }
        }
      }
      //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() );
    }
      }
      eventID_ = edm::EventID(event_->run(), currentLumiSection_, event_->event());

      setEventCached();

      return true;
    }
  }
}

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

private

Definition at line 522 of file FedRawDataInputSource.cc.

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

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

{
  const boost::filesystem::path filePath(fileName);
  if (!testModeNoBuilderUnit_) {
    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*/}
    }
  } else {
    renameToNextFree(fileName);
  }
}

bool FedRawDataInputSource::exceptionState ( )

inlineprivate

Definition at line 68 of file FedRawDataInputSource.h.

References setExceptionState_.

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

{return setExceptionState_;}

edm::Timestamp FedRawDataInputSource::fillFEDRawDataCollection ( std::auto_ptr< FEDRawDataCollection > &  rawData ) const

private

Definition at line 604 of file FedRawDataInputSource.cc.

References FEDRawData::data(), event(), event_, fedt_struct::eventsize, evf::evtn::evm_board_setformat(), FED_EVSZ_EXTRACT, FED_SOID_EXTRACT, evf::evtn::getgpshigh(), evf::evtn::getgpslow(), FEDNumbering::MINTriggerGTPFEDID, FEDRawData::resize(), and fedh_struct::sourceid.

Referenced by read().

{
  edm::Timestamp tstamp;
  uint32_t eventSize = event_->eventSize();
  char* event = (char*)event_->payload();

  while (eventSize > 0) {
    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
    eventSize -= (fedSize - sizeof(fedh_t));
    const fedh_t* fedHeader = (fedh_t *) (event + eventSize);
    const uint16_t fedId = FED_SOID_EXTRACT(fedHeader->sourceid);
    if (fedId == FEDNumbering::MINTriggerGTPFEDID) {
      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));
    }
    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 289 of file FedRawDataInputSource.cc.

References InputFile::advance(), bufferInputRead_, InputFile::bufferPosition_, checkEvery_, chunkIsFree_, InputFile::chunkPosition_, InputFile::chunks_, InputFile::currentChunk_, currentFile_, currentFileIndex_, currentLumiSection_, cvWakeup_, daqDirector_, deleteFile(), event_, eventChunkSize_, eventsThisLumi_, edm::hlt::Exception, exceptionState(), fileDeleteLock_, InputFile::fileName_, fileQueue_, InputFile::fileSize_, filesToDelete_, fms_, 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_, readNextChunkIntoBuffer(), evf::FastMonitoringService::reportEventsThisLumiInSource(), InputFile::rewindChunk(), evf::EvFDaqDirector::runEnded, evf::EvFDaqDirector::sameFile, setExceptionState_, singleBufferMode_, ntuplemaker::status, InputFile::status_, streamFileTrackerPtr_, threadError(), evf::EvFDaqDirector::updateFileIndex(), verifyAdler32_, and InputFile::waitForChunk().

Referenced by nextEvent().

{
  const size_t headerSize = (4 + 1024) * sizeof(uint32); //minimal size to fit any version of FRDEventHeader

  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::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_) {
    //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_) {
    //release last chunk (it is never released elsewhere)
    freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_]);
    if (currentFile_->nEvents_!=currentFile_->nProcessed_)
    {
      throw cms::Exception("RuntimeError") 
    << "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_ < headerSize)
  {
    throw cms::Exception("FedRawDataInputSource::cacheNextEvent") <<
      "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_;
 
    if (bufferInputRead_ < headerSize)
    {
      readNextChunkIntoBuffer(currentFile_);
      //recalculate chunk position
      dataPosition = currentFile_->chunks_[0]->buf_+ currentFile_->chunkPosition_;
      if ( bufferInputRead_ < headerSize )
      {
      throw cms::Exception("FedRawDataInputSource::cacheNextEvent") <<
    "Premature end of input file while reading event header";
      }
    }

    event_.reset( new FRDEventMsgView(dataPosition) );
    if (event_->size()>eventChunkSize_) {
      throw cms::Exception("FedRawDataInputSource::nextEvent")
          << " 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()-headerSize;

    if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize)
    {
      throw cms::Exception("FedRawDataInputSource::nextEvent") <<
    "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,headerSize);

    event_.reset( new FRDEventMsgView(dataPosition) );
    if (event_->size()>eventChunkSize_) {
      throw cms::Exception("FedRawDataInputSource::nextEvent")
          << " 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()-headerSize;

    if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize)
    {
      throw cms::Exception("FedRawDataInputSource::nextEvent") <<
    "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,headerSize);
      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(headerSize);
    //copy event to a chunk start and move pointers
    chunkEnd = currentFile_->advance(dataPosition,headerSize+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 ( verifyAdler32_ && event_->version() >= 3 )
  {
    uint32_t adler = adler32(0L,Z_NULL,0);
    adler = adler32(adler,(Bytef*)event_->payload(),event_->eventSize());

    if ( adler != event_->adler32() ) {
      throw cms::Exception("FedRawDataInputSource::nextEvent") <<
        "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 632 of file FedRawDataInputSource.cc.

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

    if ( testModeNoBuilderUnit_ )
      boost::filesystem::copy(jsonSourcePath,jsonDestPath);
    else {
      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") << "grabNextFile - BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what()
                  << " - Maybe the BU run dir disappeared? Ending process with code 0...";
    _exit(-1);
  }
  catch (std::runtime_error e)
  {
    // Another process grabbed the file and NFS did not register this
    daqDirector_->unlockFULocal();
    edm::LogError("FedRawDataInputSource") << "grabNextFile - runtime Exception -: " << e.what();
  }

  catch( boost::bad_lexical_cast const& ) {
    edm::LogError("FedRawDataInputSource") << "grabNextFile - 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 240 of file FedRawDataInputSource.cc.

References 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);
        daqDirector_->unlockFULocal2();
      }
    }

    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 279 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 ( boost::shared_ptr< edm::multicore::MessageReceiverForSource >  )

overrideprivatevirtual

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

{}

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

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 555 of file FedRawDataInputSource.cc.

References printConversionInfo::aux, edm::DaqProvenanceHelper::branchDescription(), checkEvery_, chunkIsFree_, InputFile::chunks_, InputFile::currentChunk_, currentFile_, daqProvenanceHelper_, deleteFile(), edm::DaqProvenanceHelper::dummyProvenance(), eventID_, eventsThisLumi_, fileDeleteLock_, filesToDelete_, fillFEDRawDataCollection(), freeChunks_, i, edm::RawInputSource::makeEvent(), InputFile::nProcessed_, nStreams_, edm::EventAuxiliary::PhysicsTrigger, edm::InputSource::processGUID(), processHistoryID_, edm::EventPrincipal::put(), edm::EventAuxiliary::setProcessHistoryID(), and streamFileTrackerPtr_.

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

{
  std::auto_ptr<FEDRawDataCollection> rawData(new FEDRawDataCollection);
  edm::Timestamp tstamp = fillFEDRawDataCollection(rawData);

  edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, true,
                          edm::EventAuxiliary::PhysicsTrigger);
  aux.setProcessHistoryID(processHistoryID_);
  makeEvent(eventPrincipal, aux);

  edm::WrapperOwningHolder edp(new edm::Wrapper<FEDRawDataCollection>(rawData),
                               edm::Wrapper<FEDRawDataCollection>::getInterface());

  //FWCore/Sources DaqProvenanceHelper before 7_1_0_pre3
  //eventPrincipal.put(daqProvenanceHelper_.constBranchDescription_, edp,
  //                   daqProvenanceHelper_.dummyProvenance_);
  
  eventPrincipal.put(daqProvenanceHelper_.branchDescription(), 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 1069 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 767 of file FedRawDataInputSource.cc.

References InputFile::chunks_, counter, cvReader_, cvWakeup_, daqDirector_, eventChunkSize_, fileQueue_, fms_, freeChunks_, getLSFromFilename_, grabNextJsonFile(), i, InputFile, LogDebug, python.rootplot.utilities::ls(), mReader_, mWakeup_, evf::EvFDaqDirector::newFile, evf::EvFDaqDirector::newLumi, evf::EvFDaqDirector::noFile, getHLTPrescaleColumns::path, quit_threads_, InputChunk::readComplete_, InputChunk::reset(), 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())
    {
      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;

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

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

      int dbgcount=0;
      if (status == evf::EvFDaqDirector::noFile) {
    dbgcount++;
    //if (!(dbgcount%20)) edm::LogInfo("FedRawDataInputSource") << "No file for me... sleep and try again...";
    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);
        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);
      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;
        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 939 of file FedRawDataInputSource.cc.

References InputChunk::buf_, InputFile::chunks_, cvReader_, diffTreeTool::diff, end, eventChunkBlock_, mergeVDriftHistosByStation::file, InputChunk::fileIndex_, InputFile::fileName_, first, i, init, prof2calltree::last, LogDebug, mReader_, cmsPerfSuiteHarvest::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>=1)
      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);

    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::renameToNextFree ( std::string const &  fileName ) const

private

Definition at line 741 of file FedRawDataInputSource.cc.

References daqDirector_, evf::EvFDaqDirector::getJumpFilePath(), getHLTPrescaleColumns::path, and source.

Referenced by deleteFile().

{
  boost::filesystem::path source(fileName);
  boost::filesystem::path destination( daqDirector_->getJumpFilePath() );

  edm::LogInfo("FedRawDataInputSource") << "Instead of delete, RENAME -: " << fileName
                                        << " to: " << destination.string();
  boost::filesystem::rename(source,destination);
  boost::filesystem::rename(source.replace_extension(".jsn"),destination.replace_extension(".jsn"));
}

void FedRawDataInputSource::rewind_ ( )

overrideprivatevirtual

Reimplemented from edm::RawInputSource.

Definition at line 763 of file FedRawDataInputSource.cc.

{}

void FedRawDataInputSource::threadError ( )

private

Definition at line 1009 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 42 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

friend class InputFile

friend

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

Definition at line 41 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

Member Data Documentation

uint32_t FedRawDataInputSource::bufferInputRead_ = 0

private

Definition at line 152 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readNextChunkIntoBuffer().

unsigned int FedRawDataInputSource::checkEvery_ = 10

private

Definition at line 143 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

bool FedRawDataInputSource::chunkIsFree_ =false

private

Definition at line 116 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

InputFile* FedRawDataInputSource::currentFile_ = nullptr

private

Definition at line 115 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

int FedRawDataInputSource::currentFileIndex_ = -1

private

Definition at line 137 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

unsigned int FedRawDataInputSource::currentLumiSection_

private

Definition at line 101 of file FedRawDataInputSource.h.

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

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

private

Definition at line 129 of file FedRawDataInputSource.h.

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

std::condition_variable FedRawDataInputSource::cvWakeup_

private

Definition at line 147 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

evf::EvFDaqDirector* FedRawDataInputSource::daqDirector_ =nullptr

private

Definition at line 75 of file FedRawDataInputSource.h.

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

const edm::DaqProvenanceHelper FedRawDataInputSource::daqProvenanceHelper_

private

Definition at line 94 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and read().

std::string FedRawDataInputSource::defPath_

private

Definition at line 77 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

jsoncollector::DataPointDefinition* FedRawDataInputSource::dpd_

private

Definition at line 105 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and grabNextJsonFile().

std::unique_ptr<FRDEventMsgView> FedRawDataInputSource::event_

private

Definition at line 96 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::eventChunkBlock_

private

Definition at line 80 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::eventChunkSize_

private

Definition at line 79 of file FedRawDataInputSource.h.

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

edm::EventID FedRawDataInputSource::eventID_

private

Definition at line 98 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and read().

unsigned int FedRawDataInputSource::eventsThisLumi_

private

Definition at line 102 of file FedRawDataInputSource.h.

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

unsigned long FedRawDataInputSource::eventsThisRun_ = 0

private

Definition at line 103 of file FedRawDataInputSource.h.

std::mutex FedRawDataInputSource::fileDeleteLock_

private

Definition at line 140 of file FedRawDataInputSource.h.

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

int FedRawDataInputSource::fileDescriptor_ = -1

private

Definition at line 151 of file FedRawDataInputSource.h.

Referenced by readNextChunkIntoBuffer().

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

private

Definition at line 139 of file FedRawDataInputSource.h.

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

private

Definition at line 126 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

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

private

Definition at line 138 of file FedRawDataInputSource.h.

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

evf::FastMonitoringService* FedRawDataInputSource::fms_ =nullptr

private

Definition at line 74 of file FedRawDataInputSource.h.

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

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

private

Definition at line 125 of file FedRawDataInputSource.h.

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

const std::string FedRawDataInputSource::fuOutputDir_

private

Definition at line 92 of file FedRawDataInputSource.h.

Referenced by grabNextJsonFile().

const bool FedRawDataInputSource::getLSFromFilename_

private

Definition at line 86 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::mReader_

private

Definition at line 128 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::mWakeup_

private

Definition at line 146 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

unsigned int FedRawDataInputSource::nStreams_ = 0

private

Definition at line 142 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

unsigned int FedRawDataInputSource::numBuffers_

private

Definition at line 82 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

unsigned int FedRawDataInputSource::numConcurrentReads_

private

Definition at line 83 of file FedRawDataInputSource.h.

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

edm::ProcessHistoryID FedRawDataInputSource::processHistoryID_

private

Definition at line 99 of file FedRawDataInputSource.h.

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

std::atomic<bool> FedRawDataInputSource::quit_threads_

private

Definition at line 131 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::readBlocks_

private

Definition at line 81 of file FedRawDataInputSource.h.

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

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

private

Definition at line 119 of file FedRawDataInputSource.h.

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

const edm::RunNumber_t FedRawDataInputSource::runNumber_

private

Definition at line 90 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and postForkReacquireResources().

bool FedRawDataInputSource::setExceptionState_ = false

private

Definition at line 133 of file FedRawDataInputSource.h.

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

bool FedRawDataInputSource::singleBufferMode_

private

Definition at line 150 of file FedRawDataInputSource.h.

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

bool FedRawDataInputSource::startedSupervisorThread_ = false

private

Definition at line 118 of file FedRawDataInputSource.h.

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

std::condition_variable FedRawDataInputSource::startupCv_

private

Definition at line 135 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::startupLock_

private

Definition at line 134 of file FedRawDataInputSource.h.

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

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

private

Definition at line 141 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

const bool FedRawDataInputSource::testModeNoBuilderUnit_

private

Definition at line 88 of file FedRawDataInputSource.h.

Referenced by deleteFile(), FedRawDataInputSource(), and grabNextJsonFile().

std::vector<bool> FedRawDataInputSource::thread_quit_signal

private

Definition at line 132 of file FedRawDataInputSource.h.

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

std::atomic<bool> FedRawDataInputSource::threadInit_

private

Definition at line 154 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and readWorker().

const bool FedRawDataInputSource::verifyAdler32_

private

Definition at line 87 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

std::vector<ReaderInfo> FedRawDataInputSource::workerJob_

private

Definition at line 123 of file FedRawDataInputSource.h.

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

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

private

Definition at line 122 of file FedRawDataInputSource.h.

Referenced by readSupervisor(), and readWorker().

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

private

Definition at line 120 of file FedRawDataInputSource.h.

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