FedRawDataInputSource Class Reference

#include <FedRawDataInputSource.h>

Inheritance diagram for FedRawDataInputSource:

Public Member Functions
	FedRawDataInputSource (edm::ParameterSet const &, edm::InputSourceDescription const &)
std::pair< bool, unsigned int >	getEventReport (unsigned int lumi, bool erase)
virtual	~FedRawDataInputSource ()
Public Member Functions inherited from edm::RawInputSource
	RawInputSource (ParameterSet const &pset, InputSourceDescription const &desc)
virtual	~RawInputSource ()
Public Member Functions inherited from edm::InputSource
std::shared_ptr< ActivityRegistry >	actReg () const
	Accessor for Activity Registry. More...
std::shared_ptr< BranchIDListHelper const >	branchIDListHelper () const
	Accessors for branchIDListHelper. More...
std::shared_ptr< BranchIDListHelper > &	branchIDListHelper ()
void	closeFile (FileBlock *, bool cleaningUpAfterException)
	close current file More...
void	doBeginJob ()
	Called by framework at beginning of job. More...
void	doBeginLumi (LuminosityBlockPrincipal &lbp, ProcessContext const *)
	Called by framework at beginning of lumi block. More...
void	doBeginRun (RunPrincipal &rp, ProcessContext const *)
	Called by framework at beginning of run. More...
void	doEndJob ()
	Called by framework at end of job. More...
void	doEndLumi (LuminosityBlockPrincipal &lbp, bool cleaningUpAfterException, ProcessContext const *)
	Called by framework at end of lumi block. More...
void	doEndRun (RunPrincipal &rp, bool cleaningUpAfterException, ProcessContext const *)
	Called by framework at end of run. More...
void	doPostForkReacquireResources (std::shared_ptr< multicore::MessageReceiverForSource >)
void	doPreForkReleaseResources ()
	Called by the framework before forking the process. More...
ProcessingController::ForwardState	forwardState () const
bool	goToEvent (EventID const &eventID)
	InputSource (ParameterSet const &, InputSourceDescription const &)
	Constructor. More...
	InputSource (InputSource const &)=delete
void	issueReports (EventID const &eventID)
	issue an event report More...
LuminosityBlockNumber_t	luminosityBlock () const
	Accessor for current luminosity block number. More...
std::shared_ptr< LuminosityBlockAuxiliary >	luminosityBlockAuxiliary () const
	Called by the framework to merge or insert lumi in principal cache. More...
int	maxEvents () const
int	maxLuminosityBlocks () const
ModuleDescription const &	moduleDescription () const
	Accessor for 'module' description. More...
ItemType	nextItemType ()
	Advances the source to the next item. More...
InputSource &	operator= (InputSource const &)=delete
ProcessConfiguration const &	processConfiguration () const
	Accessor for Process Configuration. More...
std::string const &	processGUID () const
	Accessor for global process identifier. More...
ProcessHistoryRegistry const &	processHistoryRegistry () const
	Accessors for process history registry. More...
ProcessHistoryRegistry &	processHistoryRegistry ()
ProcessingMode	processingMode () const
	RunsLumisAndEvents (default), RunsAndLumis, or Runs. More...
std::shared_ptr< ProductRegistry const >	productRegistry () const
	Accessors for product registry. More...
std::shared_ptr< ProductRegistry > &	productRegistry ()
bool	randomAccess () const
void	readAndMergeLumi (LuminosityBlockPrincipal &lbp)
	Read next luminosity block (same as a prior lumi) More...
void	readAndMergeRun (RunPrincipal &rp)
	Read next run (same as a prior run) More...
void	readEvent (EventPrincipal &ep, StreamContext &)
	Read next event. More...
bool	readEvent (EventPrincipal &ep, EventID const &, StreamContext &)
	Read a specific event. More...
std::unique_ptr< FileBlock >	readFile ()
	Read next file. More...
void	readLuminosityBlock (LuminosityBlockPrincipal &lumiPrincipal, HistoryAppender &historyAppender)
	Read next luminosity block (new lumi) More...
std::shared_ptr< LuminosityBlockAuxiliary >	readLuminosityBlockAuxiliary ()
	Read next luminosity block Auxilary. More...
void	readRun (RunPrincipal &runPrincipal, HistoryAppender &historyAppender)
	Read next run (new run) More...
std::shared_ptr< RunAuxiliary >	readRunAuxiliary ()
	Read next run Auxiliary. More...
ProcessHistoryID const &	reducedProcessHistoryID () const
void	registerProducts ()
	Register any produced products. More...
int	remainingEvents () const
int	remainingLuminosityBlocks () const
void	repeat ()
	Reset the remaining number of events/lumis to the maximum number. More...
std::pair< SharedResourcesAcquirer *, std::recursive_mutex * >	resourceSharedWithDelayedReader ()
	Returns nullptr if no resource shared between the Source and a DelayedReader. More...
ProcessingController::ReverseState	reverseState () const
void	rewind ()
	Begin again at the first event. More...
RunNumber_t	run () const
	Accessor for current run number. More...
std::shared_ptr< RunAuxiliary >	runAuxiliary () const
	Called by the framework to merge or insert run in principal cache. More...
void	setLuminosityBlockNumber_t (LuminosityBlockNumber_t lb)
	Set the luminosity block ID. More...
void	setRunNumber (RunNumber_t r)
	Set the run number. More...
void	skipEvents (int offset)
bool	skipForForking ()
std::shared_ptr< ThinnedAssociationsHelper const >	thinnedAssociationsHelper () const
	Accessors for thinnedAssociationsHelper. More...
std::shared_ptr< ThinnedAssociationsHelper > &	thinnedAssociationsHelper ()
Timestamp const &	timestamp () const
	Accessor for the current time, as seen by the input source. More...
virtual	~InputSource () noexcept(false)
	Destructor. More...

Static Public Member Functions
static void	fillDescriptions (edm::ConfigurationDescriptions &descriptions)
Static Public Member Functions inherited from edm::RawInputSource
static void	fillDescription (ParameterSetDescription &description)
Static Public Member Functions inherited from edm::InputSource
static const std::string &	baseType ()
static void	fillDescription (ParameterSetDescription &desc)
static void	fillDescriptions (ConfigurationDescriptions &descriptions)
static void	prevalidate (ConfigurationDescriptions &)

Protected Member Functions
virtual bool	checkNextEvent () override
virtual void	read (edm::EventPrincipal &eventPrincipal) override
Protected Member Functions inherited from edm::RawInputSource
void	makeEvent (EventPrincipal &eventPrincipal, EventAuxiliary const &eventAuxiliary)
void	setInputFileTransitionsEachEvent ()
Protected Member Functions inherited from edm::InputSource
void	decreaseRemainingEventsBy (int iSkipped)
bool	eventCached () const
bool	newLumi () const
bool	newRun () const
ProcessHistoryRegistry &	processHistoryRegistryForUpdate ()
ProductRegistry &	productRegistryUpdate ()
void	reset () const
void	resetEventCached ()
void	resetLuminosityBlockAuxiliary (bool isNewLumi=true) const
void	resetNewLumi ()
void	resetNewRun ()
void	resetRunAuxiliary (bool isNewRun=true) const
void	setEventCached ()
	Called by the framework to merge or ached() const {return eventCached_;}. More...
void	setLuminosityBlockAuxiliary (LuminosityBlockAuxiliary *lbp)
void	setNewLumi ()
void	setNewRun ()
void	setRunAuxiliary (RunAuxiliary *rp)
void	setTimestamp (Timestamp const &theTime)
	To set the current time, as seen by the input source. More...
virtual void	skip (int offset)
ItemType	state () const

Private Types
typedef std::pair< InputFile *, InputChunk * >	ReaderInfo

Private Member Functions
void	createBoLSFile (const uint32_t lumiSection, bool checkIfExists)
void	deleteFile (std::string const &)
bool	exceptionState ()
edm::Timestamp	fillFEDRawDataCollection (FEDRawDataCollection &)
evf::EvFDaqDirector::FileStatus	getFile (unsigned int &ls, std::string &nextFile, uint32_t &fsize, uint64_t &lockWaitTime)
evf::EvFDaqDirector::FileStatus	getNextEvent ()
int	grabNextJsonFile (boost::filesystem::path const &)
long	initFileList ()
void	maybeOpenNewLumiSection (const uint32_t lumiSection)
evf::EvFDaqDirector::FileStatus	nextEvent ()
virtual void	postForkReacquireResources (std::shared_ptr< edm::multicore::MessageReceiverForSource >) override
virtual void	preForkReleaseResources () override
void	readNextChunkIntoBuffer (InputFile *file)
void	readSupervisor ()
void	readWorker (unsigned int tid)
void	reportEventsThisLumiInSource (unsigned int lumi, unsigned int events)
virtual void	rewind_ () override
void	threadError ()

Private Attributes
uint32_t	bufferInputRead_ = 0
unsigned int	checkEvery_ = 10
bool	chunkIsFree_ =false
InputFile *	currentFile_ = 0
int	currentFileIndex_ = -1
unsigned int	currentLumiSection_
std::vector< std::condition_variable * >	cvReader_
std::condition_variable	cvWakeup_
evf::EvFDaqDirector *	daqDirector_ = 0
const edm::DaqProvenanceHelper	daqProvenanceHelper_
std::string	defPath_
uint32	detectedFRDversion_ =0
jsoncollector::DataPointDefinition *	dpd_
std::unique_ptr< FRDEventMsgView >	event_
unsigned int	eventChunkBlock_
unsigned int	eventChunkSize_
edm::EventID	eventID_
uint32_t	eventRunNumber_ =0
unsigned int	eventsThisLumi_
unsigned long	eventsThisRun_ = 0
std::mutex	fileDeleteLock_
int	fileDescriptor_ = -1
unsigned int	fileListIndex_ = 0
const bool	fileListLoopMode_
const bool	fileListMode_
std::vector< std::string >	fileNames_
std::list< std::pair< int, std::string > >	fileNamesToDelete_
tbb::concurrent_queue< InputFile * >	fileQueue_
std::list< std::pair< int, InputFile * > >	filesToDelete_
evf::FastMonitoringService *	fms_ = 0
tbb::concurrent_queue< InputChunk * >	freeChunks_
std::string	fuOutputDir_
const bool	getLSFromFilename_
uint32_t	GTPEventID_ = 0
uint32_t	L1EventID_ = 0
unsigned int	loopModeIterationInc_ = 0
unsigned int	maxBufferedFiles_
std::mutex	monlock_
std::mutex	mReader_
std::mutex	mWakeup_
unsigned int	nStreams_ = 0
unsigned int	numBuffers_
unsigned int	numConcurrentReads_
edm::ProcessHistoryID	processHistoryID_
std::atomic< bool >	quit_threads_
unsigned int	readBlocks_
std::atomic< unsigned int >	readingFilesCount_
std::unique_ptr< std::thread >	readSupervisorThread_
edm::RunNumber_t	runNumber_
bool	setExceptionState_ = false
bool	singleBufferMode_
std::map< unsigned int, unsigned int >	sourceEventsReport_
bool	startedSupervisorThread_ = false
std::condition_variable	startupCv_
std::mutex	startupLock_
std::vector< int >	streamFileTracker_
unsigned char *	tcds_pointer_
std::vector< unsigned int >	thread_quit_signal
std::atomic< bool >	threadInit_
std::vector< unsigned int >	tid_active_
const bool	useL1EventID_
const bool	verifyAdler32_
const bool	verifyChecksum_
std::vector< ReaderInfo >	workerJob_
tbb::concurrent_queue< unsigned int >	workerPool_
std::vector< std::thread * >	workerThreads_

Friends
struct	InputChunk
struct	InputFile

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
Public Attributes inherited from edm::InputSource
signalslot::Signal< void(StreamContext const &, ModuleCallingContext const &)>	postEventReadFromSourceSignal_
signalslot::Signal< void(StreamContext const &, ModuleCallingContext const &)>	preEventReadFromSourceSignal_

Detailed Description

Definition at line 43 of file FedRawDataInputSource.h.

Member Typedef Documentation

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

private

Definition at line 138 of file FedRawDataInputSource.h.

Constructor & Destructor Documentation

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

explicit

Definition at line 54 of file FedRawDataInputSource.cc.

References edm::Timestamp::beginOfTime(), crc32c_hw_test(), cvReader_, daqDirector_, edm::DaqProvenanceHelper::daqInit(), daqProvenanceHelper_, defPath_, dpd_, MillePedeFileConverter_cfg::e, eventChunkBlock_, eventChunkSize_, Exception, fileDeleteLock_, fileListLoopMode_, fileListMode_, filesToDelete_, fms_, freeChunks_, mps_fire::i, evf::FastMonitoringThread::inInit, initFileList(), InputChunk, edm::Timestamp::invalidTimestamp(), numBuffers_, numConcurrentReads_, Utilities::operator, processHistoryID_, edm::InputSource::processHistoryRegistryForUpdate(), edm::InputSource::productRegistryUpdate(), quit_threads_, readBlocks_, readingFilesCount_, readWorker(), edm::InputSource::runAuxiliary(), runNumber_, evf::EvFDaqDirector::setDeleteTracking(), evf::EvFDaqDirector::setFMS(), evf::FastMonitoringService::setInputSource(), evf::FastMonitoringService::setInState(), evf::FastMonitoringService::setInStateSup(), edm::InputSource::setNewRun(), edm::InputSource::setRunAuxiliary(), singleBufferMode_, startupCv_, startupLock_, trackingPlots::stat, AlCaHLTBitMon_QueryRunRegistry::string, thread_quit_signal, threadInit_, tid_active_, workerJob_, and workerThreads_.

                                                                                    :
  edm::RawInputSource(pset, desc),
  defPath_(pset.getUntrackedParameter<std::string> ("buDefPath", std::string(getenv("CMSSW_BASE"))+"/src/EventFilter/Utilities/plugins/budef.jsd")),
  eventChunkSize_(pset.getUntrackedParameter<unsigned int> ("eventChunkSize",32)*1048576),
  eventChunkBlock_(pset.getUntrackedParameter<unsigned int> ("eventChunkBlock",32)*1048576),
  numBuffers_(pset.getUntrackedParameter<unsigned int> ("numBuffers",2)),
  maxBufferedFiles_(pset.getUntrackedParameter<unsigned int> ("maxBufferedFiles",2)),
  getLSFromFilename_(pset.getUntrackedParameter<bool> ("getLSFromFilename", true)),
  verifyAdler32_(pset.getUntrackedParameter<bool> ("verifyAdler32", true)),
  verifyChecksum_(pset.getUntrackedParameter<bool> ("verifyChecksum", true)),
  useL1EventID_(pset.getUntrackedParameter<bool> ("useL1EventID", false)),
  fileNames_(pset.getUntrackedParameter<std::vector<std::string>> ("fileNames",std::vector<std::string>())),
  fileListMode_(pset.getUntrackedParameter<bool> ("fileListMode", false)),
  fileListLoopMode_(pset.getUntrackedParameter<bool> ("fileListLoopMode", false)),
  runNumber_(edm::Service<evf::EvFDaqDirector>()->getRunNumber()),
  fuOutputDir_(std::string()),
  daqProvenanceHelper_(edm::TypeID(typeid(FEDRawDataCollection))),
  eventID_(),
  processHistoryID_(),
  currentLumiSection_(0),
  tcds_pointer_(0),
  eventsThisLumi_(0),
  dpd_(nullptr)
{
  char thishost[256];
  gethostname(thishost, 255);
  edm::LogInfo("FedRawDataInputSource") << "Construction. read-ahead chunk size -: "
                                        << std::endl << (eventChunkSize_/1048576)
                                        << " MB on host " << thishost;

  long autoRunNumber = -1;
  if (fileListMode_)  {
    autoRunNumber = initFileList();
    if (!fileListLoopMode_) {
      if (autoRunNumber<0)
        throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource") << "Run number not found from filename";
      //override run number
      runNumber_ = (edm::RunNumber_t)autoRunNumber;
      edm::Service<evf::EvFDaqDirector>()->overrideRunNumber((unsigned int)autoRunNumber);
    }
  }

  processHistoryID_ = daqProvenanceHelper_.daqInit(productRegistryUpdate(), processHistoryRegistryForUpdate());
  setNewRun();
  //todo:autodetect from file name (assert if names differ)
  setRunAuxiliary(new edm::RunAuxiliary(runNumber_, edm::Timestamp::beginOfTime(),
                    edm::Timestamp::invalidTimestamp()));

  std::string defPathSuffix = "src/EventFilter/Utilities/plugins/budef.jsd";
  defPath_ = std::string(getenv("CMSSW_BASE")) + "/" + defPathSuffix;
  struct stat statbuf;
  if (stat(defPath_.c_str(), &statbuf)) {
    defPath_ = std::string(getenv("CMSSW_RELEASE_BASE")) + "/" + defPathSuffix;
    if (stat(defPath_.c_str(), &statbuf)) {
      defPath_ = defPathSuffix;
    }
  }

  dpd_ = new DataPointDefinition();
  std::string defLabel = "data";
  DataPointDefinition::getDataPointDefinitionFor(defPath_, dpd_,&defLabel);

  //make sure that chunk size is N * block size
  assert(eventChunkSize_>=eventChunkBlock_);
  readBlocks_ = eventChunkSize_/eventChunkBlock_;
  if (readBlocks_*eventChunkBlock_ != eventChunkSize_)
    eventChunkSize_=readBlocks_*eventChunkBlock_;

  if (!numBuffers_)
    throw cms::Exception("FedRawDataInputSource::FedRawDataInputSource") <<
               "no reading enabled with numBuffers parameter 0";

  numConcurrentReads_=numBuffers_-1;
  singleBufferMode_ = !(numBuffers_>1);
  readingFilesCount_=0;

  if (!crc32c_hw_test())
    edm::LogError("FedRawDataInputSource::FedRawDataInputSource") << "Intel crc32c checksum computation unavailable";

  //get handles to DaqDirector and FastMonitoringService because getting them isn't possible in readSupervisor thread
  if (fileListMode_) {
    try {
      fms_ = static_cast<evf::FastMonitoringService *> (edm::Service<evf::MicroStateService>().operator->());
    } catch(cms::Exception e) {
        edm::LogInfo("FedRawDataInputSource") << "No FastMonitoringService found in the configuration";
    }
  }
  else {
    fms_ = static_cast<evf::FastMonitoringService *> (edm::Service<evf::MicroStateService>().operator->());
    if (!fms_) {
      throw cms::Exception("FedRawDataInputSource") << "FastMonitoringService not found";
    }
  }

  daqDirector_ = edm::Service<evf::EvFDaqDirector>().operator->();
  if (!daqDirector_)
    cms::Exception("FedRawDataInputSource") << "EvFDaqDirector not found";

  //set DaqDirector to delete files in preGlobalEndLumi callback
  daqDirector_->setDeleteTracking(&fileDeleteLock_,&filesToDelete_);
  if (fms_) {
   daqDirector_->setFMS(fms_);
   fms_->setInputSource(this);
   fms_->setInState(evf::FastMonitoringThread::inInit);
   fms_->setInStateSup(evf::FastMonitoringThread::inInit);
  }
  //should delete chunks when run stops
  for (unsigned int i=0;i<numBuffers_;i++) {
    freeChunks_.push(new InputChunk(i,eventChunkSize_));
  }

  quit_threads_ = false;

  for (unsigned int i=0;i<numConcurrentReads_;i++)
  {
    std::unique_lock<std::mutex> lk(startupLock_);
    //issue a memory fence here and in threads (constructor was segfaulting without this)
    thread_quit_signal.push_back(false);
    workerJob_.push_back(ReaderInfo(nullptr,nullptr));
    cvReader_.push_back(new std::condition_variable);
    tid_active_.push_back(0);
    threadInit_.store(false,std::memory_order_release);
    workerThreads_.push_back(new std::thread(&FedRawDataInputSource::readWorker,this,i));
    startupCv_.wait(lk);
  }

  runAuxiliary()->setProcessHistoryID(processHistoryID_);
}

FedRawDataInputSource::~FedRawDataInputSource ( )

virtual

Definition at line 184 of file FedRawDataInputSource.cc.

References cvReader_, deleteFile(), filesToDelete_, mps_fire::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 234 of file FedRawDataInputSource.cc.

References evf::EvFDaqDirector::createProcessingNotificationMaybe(), currentLumiSection_, daqDirector_, evf::EvFDaqDirector::emptyLumisectionMode(), event_, eventRunNumber_, eventsThisLumi_, fileListLoopMode_, fileListMode_, fms_, runEdmFileComparison::found, fuOutputDir_, evf::EvFDaqDirector::getEoLSFilePathOnBU(), evf::EvFDaqDirector::getEoLSFilePathOnFU(), evf::EvFDaqDirector::getEoRFilePathOnFU(), getLSFromFilename_, evf::FastMonitoringThread::inWaitInput, L1EventID_, evf::EvFDaqDirector::lockFULocal2(), maybeOpenNewLumiSection(), evf::EvFDaqDirector::newLumi, nextEvent(), evf::EvFDaqDirector::noFile, readSupervisor(), readSupervisorThread_, reportEventsThisLumiInSource(), edm::InputSource::resetLuminosityBlockAuxiliary(), evf::EvFDaqDirector::runEnded, runNumber_, edm::InputSource::setEventCached(), evf::FastMonitoringService::setInState(), startedSupervisorThread_, startupCv_, startupLock_, trackingPlots::stat, AlCaHLTBitMon_QueryRunRegistry::string, and evf::EvFDaqDirector::unlockFULocal2().

{
  if (!startedSupervisorThread_)
  {
    //late init of directory variable
    fuOutputDir_=edm::Service<evf::EvFDaqDirector>()->baseRunDir();

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

      setEventCached();

      return true;
    }
  }
}

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

private

Definition at line 311 of file FedRawDataInputSource.cc.

References daqDirector_, evf::EvFDaqDirector::getBoLSFilePathOnFU(), trackingPlots::stat, 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 636 of file FedRawDataInputSource.cc.

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

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

{
  //no deletion in file list mode
  if (fileListMode_) return;

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

bool FedRawDataInputSource::exceptionState ( )

inlineprivate

Definition at line 74 of file FedRawDataInputSource.h.

References PatBasicFWLiteAnalyzer::events, FrontierConditions_GlobalTag_cff::file, ws_sso_content_reader::getFile(), eostools::ls(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by getNextEvent().

{return setExceptionState_;}

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

static

Definition at line 217 of file FedRawDataInputSource.cc.

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

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

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

private

Definition at line 743 of file FedRawDataInputSource.cc.

References FEDRawData::data(), event_, fedt_struct::eventsize, evf::evtn::evm_board_sense(), Exception, FED_EVSZ_EXTRACT, FED_SOID_EXTRACT, FEDRawDataCollection::FEDData(), 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 ntuplemaker::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;
}

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

Definition at line 1393 of file FedRawDataInputSource.cc.

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

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

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

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

private

Definition at line 1436 of file FedRawDataInputSource.cc.

References fileListIndex_, fileListLoopMode_, MillePedeFileConverter_cfg::fileName, fileNames_, loopModeIterationInc_, evf::EvFDaqDirector::newFile, callgraph::path, evf::EvFDaqDirector::runEnded, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by readSupervisor().

{
  if (fileListIndex_ < fileNames_.size()) {
          nextFile = fileNames_[fileListIndex_];
          if (nextFile.find("file://")==0) nextFile=nextFile.substr(7);
          else if (nextFile.find("file:")==0) nextFile=nextFile.substr(5);
          boost::filesystem::path fileName = nextFile;
          std::string fileStem = fileName.stem().string();
          if (fileStem.find("ls")) fileStem = fileStem.substr(fileStem.find("ls")+2);
          if (fileStem.find("_")) fileStem = fileStem.substr(0,fileStem.find("_"));

          if (!fileListLoopMode_)
            ls = boost::lexical_cast<unsigned int>(fileStem);
          else //always starting from LS 1 in loop mode
            ls = 1 + loopModeIterationInc_;

          //fsize = 0;
          //lockWaitTime = 0;
          fileListIndex_++;
          return evf::EvFDaqDirector::newFile;
  }
  else {
    if (!fileListLoopMode_)
      return evf::EvFDaqDirector::runEnded;
    else {
      //loop through files until interrupted
      loopModeIterationInc_++;
      fileListIndex_=0;
      return getFile(ls,nextFile,fsize,lockWaitTime);
    }
  }
}

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

inlineprivate

Definition at line 373 of file FedRawDataInputSource.cc.

References InputFile::advance(), bufferInputRead_, InputFile::bufferPosition_, chunkIsFree_, InputFile::chunkPosition_, InputFile::chunks_, crc32c(), InputFile::currentChunk_, currentFile_, currentFileIndex_, currentLumiSection_, cvWakeup_, daqDirector_, deleteFile(), detectedFRDversion_, event_, eventChunkSize_, eventsThisLumi_, Exception, exceptionState(), fileDeleteLock_, fileListMode_, InputFile::fileName_, fileQueue_, InputFile::fileSize_, filesToDelete_, fms_, FRDHeaderVersionSize, freeChunks_, getLSFromFilename_, evf::FastMonitoringThread::inCachedEvent, evf::FastMonitoringThread::inChecksumEvent, evf::FastMonitoringThread::inChunkReceived, evf::FastMonitoringThread::inNewLumi, evf::FastMonitoringThread::inProcessingFile, evf::FastMonitoringThread::inRunEnd, evf::FastMonitoringThread::inWaitChunk, evf::FastMonitoringThread::inWaitInput, 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_, InputFile::parent_, readingFilesCount_, readNextChunkIntoBuffer(), reportEventsThisLumiInSource(), InputFile::rewindChunk(), evf::EvFDaqDirector::runAbort, evf::EvFDaqDirector::runEnded, evf::EvFDaqDirector::sameFile, evf::FastMonitoringService::setExceptionDetected(), setExceptionState_, evf::FastMonitoringService::setInState(), singleBufferMode_, mps_update::status, InputFile::status_, threadError(), mps_check::timeout, verifyAdler32_, verifyChecksum_, and InputFile::waitForChunk().

Referenced by nextEvent().

{

  if (setExceptionState_) threadError();
  if (!currentFile_)
  {
    evf::EvFDaqDirector::FileStatus status = evf::EvFDaqDirector::noFile;
    if (fms_) fms_->setInState(evf::FastMonitoringThread::inWaitInput);
    if (!fileQueue_.try_pop(currentFile_))
    {
      //sleep until wakeup (only in single-buffer mode) or timeout
      std::unique_lock<std::mutex> lkw(mWakeup_);
      if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout || !currentFile_)
        return evf::EvFDaqDirector::noFile;
    }
    status = currentFile_->status_;
    if ( status == evf::EvFDaqDirector::runEnded)
    {
      if (fms_) fms_->setInState(evf::FastMonitoringThread::inRunEnd);
      delete currentFile_;
      currentFile_=nullptr;
      return status;
    }
    else if ( status == evf::EvFDaqDirector::runAbort)
    {
      throw cms::Exception("FedRawDataInputSource::getNextEvent") << "Run has been aborted by the input source reader thread";
    }
    else if (status == evf::EvFDaqDirector::newLumi)
    {
      if (fms_) fms_->setInState(evf::FastMonitoringThread::inNewLumi);
      if (getLSFromFilename_) {
    if (currentFile_->lumi_ > currentLumiSection_) {
          reportEventsThisLumiInSource(currentLumiSection_,eventsThisLumi_);
      eventsThisLumi_=0;
          maybeOpenNewLumiSection(currentFile_->lumi_);
    }
      }
      else {//let this be picked up from next event
        status = evf::EvFDaqDirector::noFile;
      }

      delete currentFile_;
      currentFile_=nullptr;
      return status;
    }
    else if (status == evf::EvFDaqDirector::newFile) {
      currentFileIndex_++;
    }
    else
      assert(0);
  }
  if (fms_) fms_->setInState(evf::FastMonitoringThread::inProcessingFile);

  //file is empty
  if (!currentFile_->fileSize_) {
    readingFilesCount_--;
    //try to open new lumi
    assert(currentFile_->nChunks_==0);
    if (getLSFromFilename_)
      if (currentFile_->lumi_ > currentLumiSection_) {
        reportEventsThisLumiInSource(currentLumiSection_,eventsThisLumi_);
    eventsThisLumi_=0;
        maybeOpenNewLumiSection(currentFile_->lumi_);
      }
    //immediately delete empty file
    deleteFile(currentFile_->fileName_);
    delete currentFile_;
    currentFile_=nullptr;
    return evf::EvFDaqDirector::noFile;
  }

  //file is finished
  if (currentFile_->bufferPosition_==currentFile_->fileSize_) {
    readingFilesCount_--;
    //release last chunk (it is never released elsewhere)
    freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_]);
    if (currentFile_->nEvents_>=0 && currentFile_->nEvents_!=int(currentFile_->nProcessed_))
    {
      throw cms::Exception("FedRawDataInputSource::getNextEvent")
    << "Fully processed " << currentFile_->nProcessed_ 
        << " from the file " << currentFile_->fileName_ 
    << " but according to BU JSON there should be " 
    << currentFile_->nEvents_ << " events";
    }
    //try to wake up supervisor thread which might be sleeping waiting for the free chunk
    if (singleBufferMode_) {
      std::unique_lock<std::mutex> lkw(mWakeup_);
      cvWakeup_.notify_one();
    }
    bufferInputRead_=0;
    if (!daqDirector_->isSingleStreamThread() && !fileListMode_) {
      //put the file in pending delete list;
      std::unique_lock<std::mutex> lkw(fileDeleteLock_);
      filesToDelete_.push_back(std::pair<int,InputFile*>(currentFileIndex_,currentFile_));
    }
    else {
      //in single-thread and stream jobs, events are already processed
      deleteFile(currentFile_->fileName_);
      delete currentFile_;
    }
    currentFile_=nullptr;
    return evf::EvFDaqDirector::noFile;
  }


  //file is too short
  if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < FRDHeaderVersionSize[detectedFRDversion_])
  {
    throw cms::Exception("FedRawDataInputSource::getNextEvent") <<
      "Premature end of input file while reading event header";
  }
  if (singleBufferMode_) {

    //should already be there
    if (fms_) fms_->setInState(evf::FastMonitoringThread::inWaitChunk);
    while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
      usleep(10000);
      if (currentFile_->parent_->exceptionState()) currentFile_->parent_->threadError();
    }
    if (fms_) fms_->setInState(evf::FastMonitoringThread::inChunkReceived);

    unsigned char *dataPosition = currentFile_->chunks_[0]->buf_+ currentFile_->chunkPosition_;

    //conditions when read amount is not sufficient for the header to fit
    if (!bufferInputRead_ || bufferInputRead_ < FRDHeaderVersionSize[detectedFRDversion_]
       ||  eventChunkSize_ - currentFile_->chunkPosition_ < FRDHeaderVersionSize[detectedFRDversion_])
    {
      readNextChunkIntoBuffer(currentFile_);

      if (detectedFRDversion_==0) {
        detectedFRDversion_=*((uint32*)dataPosition);
        if (detectedFRDversion_>5) 
          throw cms::Exception("FedRawDataInputSource::getNextEvent")
              << "Unknown FRD version -: " << detectedFRDversion_;
        assert(detectedFRDversion_>=1);
      }

      //recalculate chunk position
      dataPosition = currentFile_->chunks_[0]->buf_+ currentFile_->chunkPosition_;
      if ( bufferInputRead_ < FRDHeaderVersionSize[detectedFRDversion_])
      {
      throw cms::Exception("FedRawDataInputSource::getNextEvent") <<
    "Premature end of input file while reading event header";
      }
    }

    event_.reset( new FRDEventMsgView(dataPosition) );
    if (event_->size()>eventChunkSize_) {
      throw cms::Exception("FedRawDataInputSource::getNextEvent")
          << " event id:"<< event_->event()<< " lumi:" << event_->lumi()
          << " run:" << event_->run() << " of size:" << event_->size()
          << " bytes does not fit into a chunk of size:" << eventChunkSize_ << " bytes";
    }

    const uint32_t msgSize = event_->size()-FRDHeaderVersionSize[detectedFRDversion_];

    if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize)
    {
      throw cms::Exception("FedRawDataInputSource::getNextEvent") <<
    "Premature end of input file while reading event data";
    }
    if (eventChunkSize_ - currentFile_->chunkPosition_ < msgSize) {
      readNextChunkIntoBuffer(currentFile_);
      //recalculate chunk position
      dataPosition = currentFile_->chunks_[0]->buf_+ currentFile_->chunkPosition_;
      event_.reset( new FRDEventMsgView(dataPosition) );
    }
    currentFile_->bufferPosition_ += event_->size();
    currentFile_->chunkPosition_ += event_->size();
    //last chunk is released when this function is invoked next time

  }
  //multibuffer mode:
  else
  {
    //wait for the current chunk to become added to the vector
    if (fms_) fms_->setInState(evf::FastMonitoringThread::inWaitChunk);
    while (!currentFile_->waitForChunk(currentFile_->currentChunk_)) {
      usleep(10000);
      if (setExceptionState_) threadError();
    }
    if (fms_) fms_->setInState(evf::FastMonitoringThread::inChunkReceived);

    //check if header is at the boundary of two chunks
    chunkIsFree_ = false;
    unsigned char *dataPosition;

    //read header, copy it to a single chunk if necessary
    bool chunkEnd = currentFile_->advance(dataPosition,FRDHeaderVersionSize[detectedFRDversion_]);

    event_.reset( new FRDEventMsgView(dataPosition) );
    if (event_->size()>eventChunkSize_) {
      throw cms::Exception("FedRawDataInputSource::getNextEvent")
          << " event id:"<< event_->event()<< " lumi:" << event_->lumi()
          << " run:" << event_->run() << " of size:" << event_->size()
          << " bytes does not fit into a chunk of size:" << eventChunkSize_ << " bytes";
    }

    const uint32_t msgSize = event_->size()-FRDHeaderVersionSize[detectedFRDversion_];

    if (currentFile_->fileSize_ - currentFile_->bufferPosition_ < msgSize)
    {
      throw cms::Exception("FedRawDataInputSource::getNextEvent") <<
    "Premature end of input file while reading event data";
    }

    if (chunkEnd) {
      //header was at the chunk boundary, we will have to move payload as well
      currentFile_->moveToPreviousChunk(msgSize,FRDHeaderVersionSize[detectedFRDversion_]);
      chunkIsFree_ = true;
    }
    else {
      //header was contiguous, but check if payload fits the chunk
      if (eventChunkSize_ - currentFile_->chunkPosition_ < msgSize) {
    //rewind to header start position
    currentFile_->rewindChunk(FRDHeaderVersionSize[detectedFRDversion_]);
    //copy event to a chunk start and move pointers
    chunkEnd = currentFile_->advance(dataPosition,FRDHeaderVersionSize[detectedFRDversion_]+msgSize);
    assert(chunkEnd);
    chunkIsFree_=true;
    //header is moved
    event_.reset( new FRDEventMsgView(dataPosition) );
      }
      else {
    //everything is in a single chunk, only move pointers forward
    chunkEnd = currentFile_->advance(dataPosition,msgSize);
    assert(!chunkEnd);
    chunkIsFree_=false;
      }
    }
  }//end multibuffer mode
  if (fms_) fms_->setInState(evf::FastMonitoringThread::inChecksumEvent);

  if (verifyChecksum_ && event_->version() >= 5)
  {
    uint32_t crc=0;
    crc = crc32c(crc,(const unsigned char*)event_->payload(),event_->eventSize());
    if ( crc != event_->crc32c() ) {
      if (fms_) fms_->setExceptionDetected(currentLumiSection_);
      throw cms::Exception("FedRawDataInputSource::getNextEvent") <<
        "Found a wrong crc32c checksum: expected 0x" << std::hex << event_->crc32c() <<
        " but calculated 0x" << crc;
    }
  }
  else if ( verifyAdler32_ && event_->version() >= 3)
  {
    uint32_t adler = adler32(0L,Z_NULL,0);
    adler = adler32(adler,(Bytef*)event_->payload(),event_->eventSize());

    if ( adler != event_->adler32() ) {
      if (fms_) fms_->setExceptionDetected(currentLumiSection_);
      throw cms::Exception("FedRawDataInputSource::getNextEvent") <<
        "Found a wrong Adler32 checksum: expected 0x" << std::hex << event_->adler32() <<
        " but calculated 0x" << adler;
    }
  }
  if (fms_) fms_->setInState(evf::FastMonitoringThread::inCachedEvent);

  currentFile_->nProcessed_++;

  return evf::EvFDaqDirector::sameFile;
}

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

private

Definition at line 797 of file FedRawDataInputSource.cc.

References popcon2dropbox::copy(), daqDirector_, data, jsoncollector::DataPoint::deserialize(), reco::dp, dpd_, MillePedeFileConverter_cfg::e, cppFunctionSkipper::exception, Exception, fuOutputDir_, jsoncollector::DataPoint::getData(), Json::Reader::getFormatedErrorMessages(), jsoncollector::DataPointDefinition::getNames(), mps_fire::i, LogDebug, Json::Reader::parse(), callgraph::path, matplotRender::reader, MatrixUtil::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;

    std::stringstream ss;
    ss << ij.rdbuf();
    if (!reader.parse(ss.str(), deserializeRoot)) {
      edm::LogError("FedRawDataInputSource") << "Failed to deserialize JSON file -: " << jsonDestPath
                                               << "\nERROR:\n" << reader.getFormatedErrorMessages()
                                               << "CONTENT:\n" << ss.str()<<".";
      throw std::runtime_error("Cannot deserialize input JSON file");
    }

    //read BU JSON
    std::string data;
    DataPoint dp;
    dp.deserialize(deserializeRoot);
    bool success = false;
    for (unsigned int i=0;i<dpd_->getNames().size();i++) {
      if (dpd_->getNames().at(i)=="NEvents")
    if (i<dp.getData().size()) {
      data = dp.getData()[i];
      success=true;
    }
    }
    if (!success) {
      if (dp.getData().size())
    data = dp.getData()[0];
      else
    throw cms::Exception("FedRawDataInputSource::grabNextJsonFile") <<
      " error reading number of events from BU JSON -: No input value " << data;
    }
    return boost::lexical_cast<int>(data);

  }
  catch (const boost::filesystem::filesystem_error& ex)
  {
    // Input dir gone?
    daqDirector_->unlockFULocal();
    edm::LogError("FedRawDataInputSource") << "grabNextJsonFile - BOOST FILESYSTEM ERROR CAUGHT -: " << ex.what();
  }
  catch (std::runtime_error e)
  {
    // Another process grabbed the file and NFS did not register this
    daqDirector_->unlockFULocal();
    edm::LogError("FedRawDataInputSource") << "grabNextJsonFile - runtime Exception -: " << e.what();
  }

  catch( boost::bad_lexical_cast const& ) {
    edm::LogError("FedRawDataInputSource") << "grabNextJsonFile - error parsing number of events from BU JSON. "
                                           << "Input value is -: " << data;
  }

  catch (std::exception e)
  {
    // BU run directory disappeared?
    daqDirector_->unlockFULocal();
    edm::LogError("FedRawDataInputSource") << "grabNextFile - SOME OTHER EXCEPTION OCCURED!!!! -: " << e.what();
  }

  return -1;
}

long FedRawDataInputSource::initFileList ( )

private

Definition at line 1407 of file FedRawDataInputSource.cc.

References a, b, end, MillePedeFileConverter_cfg::fileName, fileNames_, callgraph::path, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FedRawDataInputSource().

{
  std::sort(fileNames_.begin(),fileNames_.end(),
            [](std::string a, std::string b) {
              if (a.rfind("/")!=std::string::npos) a=a.substr(a.rfind("/"));
              if (b.rfind("/")!=std::string::npos) b=b.substr(b.rfind("/"));
              return b > a;});

  if (fileNames_.size()) {
    //get run number from first file in the vector
    boost::filesystem::path fileName = fileNames_[0];
    std::string fileStem = fileName.stem().string();
    auto end = fileStem.find("_");
    if (fileStem.find("run")==0) {
      std::string runStr = fileStem.substr(3,end-3);
      try {
        //get long to support test run numbers < 2^32
        long rval = boost::lexical_cast<long>(runStr);
        edm::LogInfo("FedRawDataInputSource") << "Autodetected run number in fileListMode -: "<< rval;
        return rval;
      }
      catch( boost::bad_lexical_cast const& ) {
        edm::LogWarning("FedRawDataInputSource") << "Unable to autodetect run number in fileListMode from file -: "<< fileName;
      }
    }
  }
  return -1;
}

void FedRawDataInputSource::maybeOpenNewLumiSection ( const uint32_t  lumiSection )

private

Definition at line 322 of file FedRawDataInputSource.cc.

References createBoLSFile(), currentLumiSection_, daqDirector_, runEdmFileComparison::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(), trackingPlots::stat, 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 363 of file FedRawDataInputSource.cc.

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

Referenced by checkNextEvent().

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

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

overrideprivatevirtual

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

{}

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

overrideprotectedvirtual

Implements edm::RawInputSource.

Definition at line 669 of file FedRawDataInputSource.cc.

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

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

{
  if (fms_) fms_->setInState(evf::FastMonitoringThread::inReadEvent);
  std::unique_ptr<FEDRawDataCollection> rawData(new FEDRawDataCollection);
  edm::Timestamp tstamp = fillFEDRawDataCollection(*rawData);

  if (useL1EventID_){
    eventID_ = edm::EventID(eventRunNumber_, currentLumiSection_, L1EventID_);
    edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, true,
                edm::EventAuxiliary::PhysicsTrigger);
    aux.setProcessHistoryID(processHistoryID_);
    makeEvent(eventPrincipal, aux);
  }
  else if(tcds_pointer_==0){
    assert(GTPEventID_);
    eventID_ = edm::EventID(eventRunNumber_, currentLumiSection_, GTPEventID_);
    edm::EventAuxiliary aux(eventID_, processGUID(), tstamp, true,
                edm::EventAuxiliary::PhysicsTrigger);
    aux.setProcessHistoryID(processHistoryID_);
    makeEvent(eventPrincipal, aux);
  }
  else{
    evf::evtn::TCDSRecord record((unsigned char *)(tcds_pointer_));
    edm::EventAuxiliary aux = evf::evtn::makeEventAuxiliary(&record,
                         eventRunNumber_,currentLumiSection_,
                                                 processGUID(),!fileListLoopMode_);
    aux.setProcessHistoryID(processHistoryID_);
    makeEvent(eventPrincipal, aux);
  }



  std::unique_ptr<edm::WrapperBase> edp(new edm::Wrapper<FEDRawDataCollection>(std::move(rawData)));

  eventPrincipal.put(daqProvenanceHelper_.branchDescription(), std::move(edp),
                     daqProvenanceHelper_.dummyProvenance());

  eventsThisLumi_++;
  if (fms_) fms_->setInState(evf::FastMonitoringThread::inReadCleanup);

  //resize vector if needed
  while (streamFileTracker_.size() <= eventPrincipal.streamID())
      streamFileTracker_.push_back(-1);

  streamFileTracker_[eventPrincipal.streamID()]=currentFileIndex_;

  //this old file check runs no more often than every 10 events
  if (!((currentFile_->nProcessed_-1)%(checkEvery_))) {
    //delete files that are not in processing
    std::unique_lock<std::mutex> lkw(fileDeleteLock_);
    auto it = filesToDelete_.begin();
    while (it!=filesToDelete_.end()) {
      bool fileIsBeingProcessed = false;
      for (unsigned int i=0;i<nStreams_;i++) {
    if (it->first == streamFileTracker_.at(i)) {
        fileIsBeingProcessed = true;
        break;
    }
      }
      if (!fileIsBeingProcessed) {
        deleteFile(it->second->fileName_);
        delete it->second;
    it = filesToDelete_.erase(it);
      }
      else it++;
    }

  }
  if (chunkIsFree_) freeChunks_.push(currentFile_->chunks_[currentFile_->currentChunk_-1]);
  chunkIsFree_=false;
  if (fms_) fms_->setInState(evf::FastMonitoringThread::inNoRequest);
  return;
}

void FedRawDataInputSource::readNextChunkIntoBuffer ( InputFile *  file )

private

Definition at line 1328 of file FedRawDataInputSource.cc.

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

Referenced by getNextEvent().

{

  if (fileDescriptor_<0) {
    fileDescriptor_ = open(file->fileName_.c_str(), O_RDONLY);
    bufferInputRead_ = 0;
    //off_t pos = lseek(fileDescriptor,0,SEEK_SET);
    if (fileDescriptor_>=0)
      LogDebug("FedRawDataInputSource") << "opened file -: " << std::endl << file->fileName_;
    else
    {
      throw cms::Exception("FedRawDataInputSource:readNextChunkIntoBuffer") << "failed to open file " << std::endl
            << file->fileName_ << " fd:" << fileDescriptor_;
    }
  }

  if (file->chunkPosition_ == 0) { //in the rare case the last byte barely fit
    uint32_t existingSize = 0;
    for (unsigned int i=0;i<readBlocks_;i++)
    {
      const ssize_t last = ::read(fileDescriptor_,( void*) (file->chunks_[0]->buf_ + existingSize), eventChunkBlock_);
      bufferInputRead_+=last;
      existingSize+=last;
    }
  }
  else {
    const uint32_t chunksize = file->chunkPosition_;
    const uint32_t blockcount=chunksize/eventChunkBlock_;
    const uint32_t leftsize = chunksize%eventChunkBlock_;
    uint32_t existingSize = eventChunkSize_ - file->chunkPosition_;
    memmove((void*) file->chunks_[0]->buf_, file->chunks_[0]->buf_ + file->chunkPosition_, existingSize);

    for (uint32_t i=0;i<blockcount;i++) {
      const ssize_t last = ::read(fileDescriptor_,( void*) (file->chunks_[0]->buf_ + existingSize), eventChunkBlock_);
      bufferInputRead_+=last;
      existingSize+=last;
    }
    if (leftsize) {
      const ssize_t last = ::read(fileDescriptor_,( void*)( file->chunks_[0]->buf_ + existingSize ), leftsize);
      bufferInputRead_+=last;
    }
    file->chunkPosition_=0;//data was moved to beginning of the chunk
  }
  if (bufferInputRead_ == file->fileSize_) { // no more data in this file
    if (fileDescriptor_!=-1)
    {
      LogDebug("FedRawDataInputSource") << "Closing input file -: " << std::endl << file->fileName_;
      close(fileDescriptor_);
      fileDescriptor_=-1;
    }
  }
}

void FedRawDataInputSource::readSupervisor ( )

private

Definition at line 918 of file FedRawDataInputSource.cc.

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

Referenced by checkNextEvent().

{
  bool stop=false;
  unsigned int currentLumiSection = 0;
  //threadInit_.exchange(true,std::memory_order_acquire);

  {
    std::unique_lock<std::mutex> lk(startupLock_);
    startupCv_.notify_one();
  }

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

  while (!stop) {

    //wait for at least one free thread and chunk
    int counter=0;
    while ((workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty() || readingFilesCount_>=maxBufferedFiles_)
    {
      //report state to monitoring
      if (fms_) {
        bool copy_active=false;
        for (auto j : tid_active_) if (j) copy_active=true;
        if (readingFilesCount_>=maxBufferedFiles_) fms_->setInStateSup(evf::FastMonitoringThread::inSupFileLimit);
        else if (freeChunks_.empty()) {
          if (copy_active)
            fms_->setInStateSup(evf::FastMonitoringThread::inSupWaitFreeChunkCopying);
          else
            fms_->setInStateSup(evf::FastMonitoringThread::inSupWaitFreeChunk);
        }
        else {
          if (copy_active)
            fms_->setInStateSup(evf::FastMonitoringThread::inSupWaitFreeThreadCopying);
          else
            fms_->setInStateSup(evf::FastMonitoringThread::inSupWaitFreeThread);
        }
      }
      std::unique_lock<std::mutex> lkw(mWakeup_);
      //sleep until woken up by condition or a timeout
      if (cvWakeup_.wait_for(lkw, std::chrono::milliseconds(100)) == std::cv_status::timeout) {
        counter++;
        //if (!(counter%50)) edm::LogInfo("FedRawDataInputSource") << "No free chunks or threads...";
        LogDebug("FedRawDataInputSource") << "No free chunks or threads...";
      }
      else {
        assert(!(workerPool_.empty() && !singleBufferMode_) || freeChunks_.empty());
      }
      if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {stop=true;break;}
    }

    if (stop) break;

    //look for a new file
    std::string nextFile;
    uint32_t fileSize;

    if (fms_) {
      fms_->setInStateSup(evf::FastMonitoringThread::inSupBusy);
      fms_->startedLookingForFile();
      fms_->setInStateSup(evf::FastMonitoringThread::inSupLockPolling);
    }

    evf::EvFDaqDirector::FileStatus status =  evf::EvFDaqDirector::noFile;

    while (status == evf::EvFDaqDirector::noFile) {
      if (quit_threads_.load(std::memory_order_relaxed) || edm::shutdown_flag.load(std::memory_order_relaxed)) {
    stop=true;
    break;
      }
     
      uint64_t thisLockWaitTimeUs=0.;
      if (fileListMode_) {
        //return LS if LS not set, otherwise return file
        status = getFile(ls,nextFile,fileSize,thisLockWaitTimeUs);
      }
      else
        status = daqDirector_->updateFuLock(ls,nextFile,fileSize,thisLockWaitTimeUs);
        
      if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inSupBusy);

      if (currentLumiSection!=ls && status==evf::EvFDaqDirector::runEnded) status=evf::EvFDaqDirector::noFile;

      //monitoring of lock wait time
      if (thisLockWaitTimeUs>0.)
        sumLockWaitTimeUs+=thisLockWaitTimeUs;
      lockCount++;
      if (ls>monLS) {
          monLS=ls;
          if (lockCount)
            if (fms_) fms_->reportLockWait(monLS,sumLockWaitTimeUs,lockCount);
          lockCount=0;
          sumLockWaitTimeUs=0;
      }

      //check again for any remaining index/EoLS files after EoR file is seen
      if ( status == evf::EvFDaqDirector::runEnded && !fileListMode_) {
        if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inRunEnd);
        usleep(100000);
        //now all files should have appeared in ramdisk, check again if any raw files were left behind
        status = daqDirector_->updateFuLock(ls,nextFile,fileSize,thisLockWaitTimeUs);
        if (currentLumiSection!=ls && status==evf::EvFDaqDirector::runEnded) status=evf::EvFDaqDirector::noFile;
      }

      if ( status == evf::EvFDaqDirector::runEnded) {
    fileQueue_.push(new InputFile(evf::EvFDaqDirector::runEnded));
    stop=true;
    break;
      }

      //queue new lumisection
      if( getLSFromFilename_ && ls > currentLumiSection) {
        //fms_->setInStateSup(evf::FastMonitoringThread::inSupNewLumi);
    currentLumiSection = ls;
    fileQueue_.push(new InputFile(evf::EvFDaqDirector::newLumi, currentLumiSection));
      }

      if( getLSFromFilename_ && currentLumiSection>0 && ls < currentLumiSection) {
          edm::LogError("FedRawDataInputSource") << "Got old LS ("<<ls<<") file from EvFDAQDirector! Expected LS:" << currentLumiSection<<". Aborting execution."<<std::endl;
      fileQueue_.push(new InputFile(evf::EvFDaqDirector::runAbort, 0));
      stop=true;
      break;
      }

      int dbgcount=0;
      if (status == evf::EvFDaqDirector::noFile) {
        if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inSupNoFile);
    dbgcount++;
    if (!(dbgcount%20)) LogDebug("FedRawDataInputSource") << "No file for me... sleep and try again...";
    usleep(100000);
      }
    }
    if ( status == evf::EvFDaqDirector::newFile ) {
      if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFile);
      LogDebug("FedRawDataInputSource") << "The director says to grab -: " << nextFile;


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

      struct stat st;
      int stat_res = stat(rawFile.c_str(),&st);
      if (stat_res==-1) {
        edm::LogError("FedRawDataInputSource") << "Can not stat file (" << errno << "):-"<< rawFile << std::endl;
        setExceptionState_=true;
        break;
      }
      fileSize=st.st_size;

      if (fms_) {
        fms_->setInStateSup(evf::FastMonitoringThread::inSupBusy);
        fms_->stoppedLookingForFile(ls);
        fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFile);
      }
      int eventsInNewFile;
      if (fileListMode_) {
        if (fileSize==0) eventsInNewFile=0;
        else eventsInNewFile=-1;
      }
      else {
        eventsInNewFile = grabNextJsonFile(nextFile);
        assert( eventsInNewFile>=0 );
        assert((eventsInNewFile>0) == (fileSize>0));//file without events must be empty
      }

      if (!singleBufferMode_) {
    //calculate number of needed chunks
    unsigned int neededChunks = fileSize/eventChunkSize_;
    if (fileSize%eventChunkSize_) neededChunks++;

        InputFile * newInputFile = new InputFile(evf::EvFDaqDirector::FileStatus::newFile,ls,rawFile,fileSize,neededChunks,eventsInNewFile,this);
        readingFilesCount_++;
        fileQueue_.push(newInputFile);

    for (unsigned int i=0;i<neededChunks;i++) {

          if (fms_) {
            bool copy_active=false;
            for (auto j : tid_active_) if (j) copy_active=true;
            if (copy_active)
              fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFileWaitThreadCopying);
            else
              fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFileWaitThread);
          }
      //get thread
      unsigned int newTid = 0xffffffff;
      while (!workerPool_.try_pop(newTid)) {
        usleep(100000);
      }

          if (fms_) {
            bool copy_active=false;
            for (auto j : tid_active_) if (j) copy_active=true;
            if (copy_active)
              fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFileWaitChunkCopying);
            else
              fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFileWaitChunk);
          }
      InputChunk * newChunk = nullptr;
      while (!freeChunks_.try_pop(newChunk)) {
            usleep(100000);
            if (quit_threads_.load(std::memory_order_relaxed)) break;
      }

          if (newChunk == nullptr) {
            //return unused tid if we received shutdown (nullptr chunk)
            if (newTid!=0xffffffff) workerPool_.push(newTid);
            stop = true;
            break;
          }
          if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inSupNewFile);

      std::unique_lock<std::mutex> lk(mReader_);

      unsigned int toRead = eventChunkSize_;
      if (i==neededChunks-1 && fileSize%eventChunkSize_) toRead = fileSize%eventChunkSize_;
      newChunk->reset(i*eventChunkSize_,toRead,i);

      workerJob_[newTid].first=newInputFile;
      workerJob_[newTid].second=newChunk;

      //wake up the worker thread
      cvReader_[newTid]->notify_one();
    }
      }
      else {
    if (!eventsInNewFile) {
      //still queue file for lumi update
      std::unique_lock<std::mutex> lkw(mWakeup_);
      InputFile * newInputFile = new InputFile(evf::EvFDaqDirector::FileStatus::newFile,ls,rawFile,0,0,0,this);
          readingFilesCount_++;
      fileQueue_.push(newInputFile);
      cvWakeup_.notify_one();
      break;
    }
    //in single-buffer mode put single chunk in the file and let the main thread read the file
    InputChunk * newChunk;
    //should be available immediately
    while(!freeChunks_.try_pop(newChunk)) usleep(100000);

        std::unique_lock<std::mutex> lkw(mWakeup_);

        unsigned int toRead = eventChunkSize_;
        if (fileSize%eventChunkSize_) toRead = fileSize%eventChunkSize_;
        newChunk->reset(0,toRead,0);
        newChunk->readComplete_=true;

        //push file and wakeup main thread
        InputFile * newInputFile = new InputFile(evf::EvFDaqDirector::FileStatus::newFile,ls,rawFile,fileSize,1,eventsInNewFile,this);
        newInputFile->chunks_[0]=newChunk;
        readingFilesCount_++;
        fileQueue_.push(newInputFile);
        cvWakeup_.notify_one();
      }
    }
  }
  if (fms_) fms_->setInStateSup(evf::FastMonitoringThread::inRunEnd);
  //make sure threads finish reading
  unsigned numFinishedThreads = 0;
  while (numFinishedThreads < workerThreads_.size()) {
    unsigned int tid;
    while (!workerPool_.try_pop(tid)) {usleep(10000);}
    std::unique_lock<std::mutex> lk(mReader_);
    thread_quit_signal[tid]=true;
    cvReader_[tid]->notify_one();
    numFinishedThreads++;
  }
  for (unsigned int i=0;i<workerThreads_.size();i++) {
    workerThreads_[i]->join();
    delete workerThreads_[i];
  }
}

void FedRawDataInputSource::readWorker ( unsigned int  tid )

private

Definition at line 1193 of file FedRawDataInputSource.cc.

References InputChunk::buf_, InputFile::chunks_, cvReader_, detectedFRDversion_, mps_update::diff, end, eventChunkBlock_, FrontierConditions_GlobalTag_cff::file, InputChunk::fileIndex_, InputFile::fileName_, plotBeamSpotDB::first, mps_fire::i, StGenEvtProducer_cfi::init, plotBeamSpotDB::last, LogDebug, mReader_, cmsPerfSuiteHarvest::now, InputChunk::offset_, read(), readBlocks_, InputChunk::readComplete_, edm::second(), setExceptionState_, command_line::start, startupCv_, startupLock_, thread_quit_signal, threadInit_, tid_active_, InputChunk::usedSize_, workerJob_, and workerPool_.

Referenced by FedRawDataInputSource().

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

  while (1) {

    tid_active_[tid]=false;
    std::unique_lock<std::mutex> lk(mReader_);
    workerJob_[tid].first=nullptr;
    workerJob_[tid].first=nullptr;

    assert(!thread_quit_signal[tid]);//should never get it here
    workerPool_.push(tid);

    if (init) {
      std::unique_lock<std::mutex> lk(startupLock_);
      init = false;
      startupCv_.notify_one();
    }
    cvReader_[tid]->wait(lk);

    if (thread_quit_signal[tid])  return;
    tid_active_[tid]=true;

    InputFile * file;
    InputChunk * chunk;

    assert(workerJob_[tid].first!=nullptr && workerJob_[tid].second!=nullptr);

    file = workerJob_[tid].first;
    chunk = workerJob_[tid].second;

    int fileDescriptor = open(file->fileName_.c_str(), O_RDONLY);
    off_t pos = lseek(fileDescriptor,chunk->offset_,SEEK_SET);


    if (fileDescriptor>=0)
      LogDebug("FedRawDataInputSource") << "Reader thread opened file -: TID: " << tid << " file: " << file->fileName_ << " at offset " << pos;
    else
    {
      edm::LogError("FedRawDataInputSource") <<
      "readWorker failed to open file -: " << file->fileName_ << " fd:" << fileDescriptor <<
      " or seek to offset " << chunk->offset_ << ", lseek returned:" << pos;
      setExceptionState_=true;
      return;

    }

    unsigned int bufferLeft = 0;
    auto start = std::chrono::high_resolution_clock::now();
    for (unsigned int i=0;i<readBlocks_;i++)
    {
      const ssize_t last = ::read(fileDescriptor,( void*) (chunk->buf_+bufferLeft), eventChunkBlock_);
      if ( last > 0 )
    bufferLeft+=last;
      if (last < eventChunkBlock_) {
    assert(chunk->usedSize_==i*eventChunkBlock_+last);
    break;
      }
    }
    auto end = std::chrono::high_resolution_clock::now();
    auto diff = end-start;
    std::chrono::milliseconds msec = std::chrono::duration_cast<std::chrono::milliseconds>(diff);
    LogDebug("FedRawDataInputSource") << " finished reading block -: " << (bufferLeft >> 20) << " MB" << " in " << msec.count() << " ms ("<< (bufferLeft >> 20)/double(msec.count())<<" GB/s)";
    close(fileDescriptor);

    if (detectedFRDversion_==0 && chunk->offset_==0) detectedFRDversion_=*((uint32*)chunk->buf_);
    assert(detectedFRDversion_<=5);
    chunk->readComplete_=true;//this is atomic to secure the sequential buffer fill before becoming available for processing)
    file->chunks_[chunk->fileIndex_]=chunk;//put the completed chunk in the file chunk vector at predetermined index

  }
}

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

private

Definition at line 1382 of file FedRawDataInputSource.cc.

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

Referenced by checkNextEvent(), and getNextEvent().

{

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

void FedRawDataInputSource::rewind_ ( )

overrideprivatevirtual

Reimplemented from edm::RawInputSource.

Definition at line 914 of file FedRawDataInputSource.cc.

{}

void FedRawDataInputSource::threadError ( )

private

Definition at line 1268 of file FedRawDataInputSource.cc.

References Exception, and quit_threads_.

Referenced by getNextEvent().

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

}

Friends And Related Function Documentation

friend struct InputChunk

friend

Definition at line 46 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

friend struct InputFile

friend

Definition at line 45 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

Member Data Documentation

uint32_t FedRawDataInputSource::bufferInputRead_ = 0

private

Definition at line 179 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readNextChunkIntoBuffer().

unsigned int FedRawDataInputSource::checkEvery_ = 10

private

Definition at line 170 of file FedRawDataInputSource.h.

Referenced by read().

bool FedRawDataInputSource::chunkIsFree_ =false

private

Definition at line 142 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

InputFile* FedRawDataInputSource::currentFile_ = 0

private

Definition at line 141 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

int FedRawDataInputSource::currentFileIndex_ = -1

private

Definition at line 164 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and read().

unsigned int FedRawDataInputSource::currentLumiSection_

private

Definition at line 122 of file FedRawDataInputSource.h.

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

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

private

Definition at line 155 of file FedRawDataInputSource.h.

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

std::condition_variable FedRawDataInputSource::cvWakeup_

private

Definition at line 174 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

evf::EvFDaqDirector* FedRawDataInputSource::daqDirector_ = 0

private

Definition at line 87 of file FedRawDataInputSource.h.

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

const edm::DaqProvenanceHelper FedRawDataInputSource::daqProvenanceHelper_

private

Definition at line 115 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and read().

std::string FedRawDataInputSource::defPath_

private

Definition at line 89 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

uint32 FedRawDataInputSource::detectedFRDversion_ =0

private

Definition at line 140 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readWorker().

jsoncollector::DataPointDefinition* FedRawDataInputSource::dpd_

private

Definition at line 130 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and grabNextJsonFile().

std::unique_ptr<FRDEventMsgView> FedRawDataInputSource::event_

private

Definition at line 117 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::eventChunkBlock_

private

Definition at line 92 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::eventChunkSize_

private

Definition at line 91 of file FedRawDataInputSource.h.

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

edm::EventID FedRawDataInputSource::eventID_

private

Definition at line 119 of file FedRawDataInputSource.h.

Referenced by read().

uint32_t FedRawDataInputSource::eventRunNumber_ =0

private

Definition at line 123 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and read().

unsigned int FedRawDataInputSource::eventsThisLumi_

private

Definition at line 127 of file FedRawDataInputSource.h.

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

unsigned long FedRawDataInputSource::eventsThisRun_ = 0

private

Definition at line 128 of file FedRawDataInputSource.h.

std::mutex FedRawDataInputSource::fileDeleteLock_

private

Definition at line 167 of file FedRawDataInputSource.h.

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

int FedRawDataInputSource::fileDescriptor_ = -1

private

Definition at line 178 of file FedRawDataInputSource.h.

Referenced by readNextChunkIntoBuffer().

unsigned int FedRawDataInputSource::fileListIndex_ = 0

private

Definition at line 108 of file FedRawDataInputSource.h.

Referenced by getFile().

const bool FedRawDataInputSource::fileListLoopMode_

private

Definition at line 109 of file FedRawDataInputSource.h.

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

const bool FedRawDataInputSource::fileListMode_

private

Definition at line 107 of file FedRawDataInputSource.h.

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

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

private

Definition at line 104 of file FedRawDataInputSource.h.

Referenced by getFile(), and initFileList().

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

private

Definition at line 166 of file FedRawDataInputSource.h.

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

private

Definition at line 152 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

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

private

Definition at line 165 of file FedRawDataInputSource.h.

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

evf::FastMonitoringService* FedRawDataInputSource::fms_ = 0

private

Definition at line 86 of file FedRawDataInputSource.h.

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

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

private

Definition at line 151 of file FedRawDataInputSource.h.

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

std::string FedRawDataInputSource::fuOutputDir_

private

Definition at line 113 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and grabNextJsonFile().

const bool FedRawDataInputSource::getLSFromFilename_

private

Definition at line 100 of file FedRawDataInputSource.h.

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

uint32_t FedRawDataInputSource::GTPEventID_ = 0

private

Definition at line 124 of file FedRawDataInputSource.h.

Referenced by fillFEDRawDataCollection(), and read().

uint32_t FedRawDataInputSource::L1EventID_ = 0

private

Definition at line 125 of file FedRawDataInputSource.h.

Referenced by checkNextEvent(), and read().

unsigned int FedRawDataInputSource::loopModeIterationInc_ = 0

private

Definition at line 110 of file FedRawDataInputSource.h.

Referenced by getFile().

unsigned int FedRawDataInputSource::maxBufferedFiles_

private

Definition at line 95 of file FedRawDataInputSource.h.

Referenced by readSupervisor().

std::mutex FedRawDataInputSource::monlock_

private

Definition at line 184 of file FedRawDataInputSource.h.

Referenced by getEventReport(), and reportEventsThisLumiInSource().

std::mutex FedRawDataInputSource::mReader_

private

Definition at line 154 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::mWakeup_

private

Definition at line 173 of file FedRawDataInputSource.h.

Referenced by getNextEvent(), and readSupervisor().

unsigned int FedRawDataInputSource::nStreams_ = 0

private

Definition at line 169 of file FedRawDataInputSource.h.

Referenced by read().

unsigned int FedRawDataInputSource::numBuffers_

private

Definition at line 94 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource().

unsigned int FedRawDataInputSource::numConcurrentReads_

private

Definition at line 96 of file FedRawDataInputSource.h.

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

edm::ProcessHistoryID FedRawDataInputSource::processHistoryID_

private

Definition at line 120 of file FedRawDataInputSource.h.

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

std::atomic<bool> FedRawDataInputSource::quit_threads_

private

Definition at line 158 of file FedRawDataInputSource.h.

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

unsigned int FedRawDataInputSource::readBlocks_

private

Definition at line 93 of file FedRawDataInputSource.h.

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

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

private

Definition at line 97 of file FedRawDataInputSource.h.

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

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

private

Definition at line 145 of file FedRawDataInputSource.h.

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

edm::RunNumber_t FedRawDataInputSource::runNumber_

private

Definition at line 112 of file FedRawDataInputSource.h.

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

bool FedRawDataInputSource::setExceptionState_ = false

private

Definition at line 160 of file FedRawDataInputSource.h.

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

bool FedRawDataInputSource::singleBufferMode_

private

Definition at line 177 of file FedRawDataInputSource.h.

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

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

private

Definition at line 183 of file FedRawDataInputSource.h.

Referenced by getEventReport(), and reportEventsThisLumiInSource().

bool FedRawDataInputSource::startedSupervisorThread_ = false

private

Definition at line 144 of file FedRawDataInputSource.h.

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

std::condition_variable FedRawDataInputSource::startupCv_

private

Definition at line 162 of file FedRawDataInputSource.h.

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

std::mutex FedRawDataInputSource::startupLock_

private

Definition at line 161 of file FedRawDataInputSource.h.

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

std::vector<int> FedRawDataInputSource::streamFileTracker_

private

Definition at line 168 of file FedRawDataInputSource.h.

Referenced by read().

unsigned char* FedRawDataInputSource::tcds_pointer_

private

Definition at line 126 of file FedRawDataInputSource.h.

Referenced by fillFEDRawDataCollection(), and read().

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

private

Definition at line 159 of file FedRawDataInputSource.h.

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

std::atomic<bool> FedRawDataInputSource::threadInit_

private

Definition at line 181 of file FedRawDataInputSource.h.

Referenced by FedRawDataInputSource(), and readWorker().

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

private

Definition at line 156 of file FedRawDataInputSource.h.

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

const bool FedRawDataInputSource::useL1EventID_

private

Definition at line 103 of file FedRawDataInputSource.h.

Referenced by read().

const bool FedRawDataInputSource::verifyAdler32_

private

Definition at line 101 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

const bool FedRawDataInputSource::verifyChecksum_

private

Definition at line 102 of file FedRawDataInputSource.h.

Referenced by getNextEvent().

std::vector<ReaderInfo> FedRawDataInputSource::workerJob_

private

Definition at line 149 of file FedRawDataInputSource.h.

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

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

private

Definition at line 148 of file FedRawDataInputSource.h.

Referenced by readSupervisor(), and readWorker().

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

private

Definition at line 146 of file FedRawDataInputSource.h.

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