evf::FastMonitoringService Class Reference

#include <FastMonitoringService.h>

Inheritance diagram for evf::FastMonitoringService:

Public Member Functions
void	accumulateFileSize (unsigned int lumi, unsigned long fileSize)
bool	exceptionDetected () const
	FastMonitoringService (const edm::ParameterSet &, edm::ActivityRegistry &)
bool	getAbortFlagForLumi (unsigned int lumi)
unsigned int	getEventsProcessedForLumi (unsigned int lumi, bool *abortFlag=nullptr)
std::string	getRunDirName () const
bool	isExceptionOnData (unsigned int ls)
void	jobFailure ()
std::string	makeInputLegendaJson ()
std::string	makeModuleLegendaJson ()
void	postBeginJob ()
void	postEndJob ()
void	postEvent (edm::StreamContext const &)
void	postGlobalBeginRun (edm::GlobalContext const &)
void	postGlobalEndLumi (edm::GlobalContext const &)
void	postModuleEvent (edm::StreamContext const &, edm::ModuleCallingContext const &)
void	postModuleEventAcquire (edm::StreamContext const &, edm::ModuleCallingContext const &)
void	postSourceEvent (edm::StreamID)
void	postStreamBeginLumi (edm::StreamContext const &)
void	postStreamEndLumi (edm::StreamContext const &)
void	preallocate (edm::service::SystemBounds const &)
void	preBeginJob (edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &pc)
void	preEvent (edm::StreamContext const &)
void	preGlobalBeginLumi (edm::GlobalContext const &)
void	preGlobalEarlyTermination (edm::GlobalContext const &, edm::TerminationOrigin)
void	preGlobalEndLumi (edm::GlobalContext const &)
void	preModuleBeginJob (edm::ModuleDescription const &)
void	preModuleEvent (edm::StreamContext const &, edm::ModuleCallingContext const &)
void	preModuleEventAcquire (edm::StreamContext const &, edm::ModuleCallingContext const &)
void	prePathEvent (edm::StreamContext const &, edm::PathContext const &)
void	preSourceEarlyTermination (edm::TerminationOrigin)
void	preSourceEvent (edm::StreamID)
void	preStreamBeginLumi (edm::StreamContext const &)
void	preStreamEarlyTermination (edm::StreamContext const &, edm::TerminationOrigin)
void	preStreamEndLumi (edm::StreamContext const &)
void	reportLockWait (unsigned int ls, double waitTime, unsigned int lockCount)
void	setExceptionDetected (unsigned int ls)
void	setInputSource (FedRawDataInputSource *inputSource)
void	setInputSource (DAQSource *inputSource)
void	setInState (FastMonState::InputState inputState)
void	setInStateSup (FastMonState::InputState inputState)
void	setTMicrostate (FastMonState::Microstate m)
bool	shouldWriteFiles (unsigned int lumi, unsigned int *proc=nullptr)
void	startedLookingForFile ()
void	stoppedLookingForFile (unsigned int lumi)
	~FastMonitoringService () override
Public Member Functions inherited from evf::MicroStateService
	MicroStateService (const edm::ParameterSet &, edm::ActivityRegistry &)
virtual	~MicroStateService ()

Static Public Member Functions
static void	fillDescriptions (edm::ConfigurationDescriptions &descriptions)
static unsigned int	getTID ()

Static Public Attributes
static const std::string	inputStateNames [FastMonState::inCOUNT]
static const std::string	macroStateNames [FastMonState::MCOUNT]
static const edm::ModuleDescription	specialMicroStateNames [FastMonState::mCOUNT]

Private Member Functions
void	doSnapshot (const unsigned int ls, const bool isGlobalEOL)
void	snapshotRunner ()

Static Private Member Functions
static unsigned int	getSID (edm::StreamContext const &sc)
static unsigned int	getSID (edm::StreamID const &sid)

Private Attributes
std::map< unsigned int, unsigned long >	accuSize_
std::map< unsigned int, double >	avgLeadTime_
std::unique_ptr< ConcurrencyTracker >	ct_
DAQSource *	daqInputSource_ = nullptr
bool	exception_detected_ = false
std::vector< unsigned int >	exceptionInLS_
std::string	fastMicrostateDefPath_
unsigned int	fastMonIntervals_
std::string	fastName_
std::string	fastPath_
timeval	fileLookStart_
timeval	fileLookStop_
std::map< unsigned int, unsigned int >	filesProcessedDuringLumi_
std::unique_ptr< FastMonitoringThread >	fmt_
std::atomic< bool >	has_data_exception_ = false
std::atomic< bool >	has_source_exception_ = false
std::string	inputLegendFileJson_
FedRawDataInputSource *	inputSource_ = nullptr
std::atomic< FastMonState::InputState >	inputState_ {FastMonState::InputState::inInit}
std::atomic< FastMonState::InputState >	inputSupervisorState_ {FastMonState::InputState::inInit}
std::atomic< bool >	isInitTransition_
unsigned int	lastGlobalLumi_
std::vector< double >	leadTimes_
std::map< unsigned int, std::pair< double, unsigned int > >	lockStatsDuringLumi_
unsigned int	lumiFromSource_
std::map< unsigned int, timeval >	lumiStartTime_
std::vector< ContainableAtomic< const void * > >	microstate_
std::vector< ContainableAtomic< unsigned char > >	microstateAcqFlag_
std::string	microstateDefPath_
std::string	moduleLegendFile_
std::string	moduleLegendFileJson_
std::atomic< bool >	monInit_
unsigned int	nMonThreads_ = 0
unsigned int	nOutputModules_ = 0
unsigned int	nStreams_ = 0
unsigned int	nThreads_ = 0
std::map< unsigned int, std::pair< unsigned int, bool > >	processedEventsPerLumi_
std::filesystem::path	runDirectory_
int	sleepTime_
unsigned int	snapCounter_ = 0
std::vector< std::atomic< bool > * >	streamCounterUpdating_
bool	tbbConcurrencyTracker_
bool	tbbMonitoringMode_
bool	threadIDAvailable_ = false
std::vector< ContainableAtomic< const void * > >	tmicrostate_
std::vector< ContainableAtomic< unsigned char > >	tmicrostateAcqFlag_
std::atomic< unsigned long >	totalEventsProcessed_
bool	verbose_ = false
std::filesystem::path	workingDirectory_

Detailed Description

Definition at line 167 of file FastMonitoringService.h.

Constructor & Destructor Documentation

FastMonitoringService()

evf::FastMonitoringService::FastMonitoringService	(	const edm::ParameterSet &	iPS,
		edm::ActivityRegistry &	reg
	)

Definition at line 197 of file FastMonitoringService.cc.

References Exception, fastMicrostateDefPath_, jobFailure(), microstateDefPath_, postBeginJob(), postEndJob(), postEvent(), postGlobalEndLumi(), postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), postStreamBeginLumi(), postStreamEndLumi(), preallocate(), preBeginJob(), preEvent(), preGlobalBeginLumi(), preGlobalEarlyTermination(), preGlobalEndLumi(), preModuleBeginJob(), preModuleEvent(), preModuleEventAcquire(), preSourceEarlyTermination(), preSourceEvent(), preStreamBeginLumi(), preStreamEarlyTermination(), preStreamEndLumi(), edm_modernize_messagelogger::stat, AlCaHLTBitMon_QueryRunRegistry::string, edm::ActivityRegistry::watchJobFailure(), edm::ActivityRegistry::watchPostBeginJob(), edm::ActivityRegistry::watchPostEndJob(), edm::ActivityRegistry::watchPostEvent(), edm::ActivityRegistry::watchPostGlobalEndLumi(), edm::ActivityRegistry::watchPostModuleEvent(), edm::ActivityRegistry::watchPostModuleEventAcquire(), edm::ActivityRegistry::watchPostSourceEvent(), edm::ActivityRegistry::watchPostStreamBeginLumi(), edm::ActivityRegistry::watchPostStreamEndLumi(), edm::ActivityRegistry::watchPreallocate(), edm::ActivityRegistry::watchPreBeginJob(), edm::ActivityRegistry::watchPreEvent(), edm::ActivityRegistry::watchPreGlobalBeginLumi(), edm::ActivityRegistry::watchPreGlobalEarlyTermination(), edm::ActivityRegistry::watchPreGlobalEndLumi(), edm::ActivityRegistry::watchPreModuleBeginJob(), edm::ActivityRegistry::watchPreModuleEvent(), edm::ActivityRegistry::watchPreModuleEventAcquire(), edm::ActivityRegistry::watchPreSourceEarlyTermination(), edm::ActivityRegistry::watchPreSourceEvent(), edm::ActivityRegistry::watchPreStreamBeginLumi(), edm::ActivityRegistry::watchPreStreamEarlyTermination(), and edm::ActivityRegistry::watchPreStreamEndLumi().

      : MicroStateService(iPS, reg),
        fmt_(new FastMonitoringThread()),
        tbbMonitoringMode_(iPS.getUntrackedParameter<bool>("tbbMonitoringMode", true)),
        tbbConcurrencyTracker_(iPS.getUntrackedParameter<bool>("tbbConcurrencyTracker", true) && tbbMonitoringMode_),
        sleepTime_(iPS.getUntrackedParameter<int>("sleepTime", 1)),
        fastMonIntervals_(iPS.getUntrackedParameter<unsigned int>("fastMonIntervals", 2)),
        fastName_("fastmoni"),
        totalEventsProcessed_(0),
        verbose_(iPS.getUntrackedParameter<bool>("verbose")) {
    reg.watchPreallocate(this, &FastMonitoringService::preallocate);  //receiving information on number of threads
    reg.watchJobFailure(this, &FastMonitoringService::jobFailure);    //global

    reg.watchPreBeginJob(this, &FastMonitoringService::preBeginJob);
    reg.watchPreModuleBeginJob(this, &FastMonitoringService::preModuleBeginJob);  //global
    reg.watchPostBeginJob(this, &FastMonitoringService::postBeginJob);
    reg.watchPostEndJob(this, &FastMonitoringService::postEndJob);

    reg.watchPreGlobalBeginLumi(this, &FastMonitoringService::preGlobalBeginLumi);  //global lumi
    reg.watchPreGlobalEndLumi(this, &FastMonitoringService::preGlobalEndLumi);
    reg.watchPostGlobalEndLumi(this, &FastMonitoringService::postGlobalEndLumi);

    reg.watchPreStreamBeginLumi(this, &FastMonitoringService::preStreamBeginLumi);  //stream lumi
    reg.watchPostStreamBeginLumi(this, &FastMonitoringService::postStreamBeginLumi);
    reg.watchPreStreamEndLumi(this, &FastMonitoringService::preStreamEndLumi);
    reg.watchPostStreamEndLumi(this, &FastMonitoringService::postStreamEndLumi);

    reg.watchPreEvent(this, &FastMonitoringService::preEvent);  //stream
    reg.watchPostEvent(this, &FastMonitoringService::postEvent);

    //readEvent (not getNextItemType)
    reg.watchPreSourceEvent(this, &FastMonitoringService::preSourceEvent);  //source (with streamID of requestor)
    reg.watchPostSourceEvent(this, &FastMonitoringService::postSourceEvent);

    reg.watchPreModuleEventAcquire(this, &FastMonitoringService::preModuleEventAcquire);  //stream
    reg.watchPostModuleEventAcquire(this, &FastMonitoringService::postModuleEventAcquire);

    reg.watchPreModuleEvent(this, &FastMonitoringService::preModuleEvent);  //stream
    reg.watchPostModuleEvent(this, &FastMonitoringService::postModuleEvent);

    reg.watchPreStreamEarlyTermination(this, &FastMonitoringService::preStreamEarlyTermination);
    reg.watchPreGlobalEarlyTermination(this, &FastMonitoringService::preGlobalEarlyTermination);
    reg.watchPreSourceEarlyTermination(this, &FastMonitoringService::preSourceEarlyTermination);

    //find microstate definition path (required by the module)
    struct stat statbuf;
    std::string microstateBaseSuffix = "src/EventFilter/Utilities/plugins/microstatedef.jsd";
    std::string microstatePath = std::string(std::getenv("CMSSW_BASE")) + "/" + microstateBaseSuffix;
    if (stat(microstatePath.c_str(), &statbuf)) {
      microstatePath = std::string(std::getenv("CMSSW_RELEASE_BASE")) + "/" + microstateBaseSuffix;
      if (stat(microstatePath.c_str(), &statbuf)) {
        microstatePath = microstateBaseSuffix;
        if (stat(microstatePath.c_str(), &statbuf))
          throw cms::Exception("FastMonitoringService") << "microstate definition file not found";
      }
    }
    fastMicrostateDefPath_ = microstateDefPath_ = microstatePath;
  }

~FastMonitoringService()

evf::FastMonitoringService::~FastMonitoringService ( )

override

Definition at line 256 of file FastMonitoringService.cc.

{}

Member Function Documentation

accumulateFileSize()

void evf::FastMonitoringService::accumulateFileSize	(	unsigned int	lumi,
		unsigned long	fileSize
	)

Definition at line 717 of file FastMonitoringService.cc.

References accuSize_, filesProcessedDuringLumi_, fmt_, CommonMethods::lock(), and BXlumiParameters_cfi::lumi.

Referenced by evf::EvFDaqDirector::bumpFile().

                                                                                          {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);

    if (accuSize_.find(lumi) == accuSize_.end())
      accuSize_[lumi] = fileSize;
    else
      accuSize_[lumi] += fileSize;

    if (filesProcessedDuringLumi_.find(lumi) == filesProcessedDuringLumi_.end())
      filesProcessedDuringLumi_[lumi] = 1;
    else
      filesProcessedDuringLumi_[lumi]++;
  }

doSnapshot()

void evf::FastMonitoringService::doSnapshot ( const unsigned int  ls, const bool  isGlobalEOL  )

private

Definition at line 859 of file FastMonitoringService.cc.

References avgLeadTime_, ct_, filesProcessedDuringLumi_, fmt_, evf::getmEoL(), evf::getmFwk(), evf::getmFwkEoL(), evf::getmIdle(), mps_fire::i, ALPAKA_ACCELERATOR_NAMESPACE::caPixelDoublets::if(), evf::FastMonState::inIgnore, evf::FastMonState::inNewLumi, evf::FastMonState::inNoRequest, evf::FastMonState::inNoRequestWithEoLThreads, evf::FastMonState::inNoRequestWithIdleThreads, inputState_, inputSupervisorState_, evf::FastMonState::inRunEnd, evf::FastMonState::inSupBusy, evf::FastMonState::inSupFileLimit, evf::FastMonState::inSupLockPolling, evf::FastMonState::inSupLockPollingCopying, evf::FastMonState::inSupNewFile, evf::FastMonState::inSupNewFileWaitChunk, evf::FastMonState::inSupNewFileWaitChunkCopying, evf::FastMonState::inSupNewFileWaitThread, evf::FastMonState::inSupNewFileWaitThreadCopying, evf::FastMonState::inSupNoFile, evf::FastMonState::inSupThrottled, evf::FastMonState::inSupWaitFreeChunk, evf::FastMonState::inSupWaitFreeChunkCopying, evf::FastMonState::inSupWaitFreeThread, evf::FastMonState::inSupWaitFreeThreadCopying, evf::FastMonState::inWaitChunk, evf::FastMonState::inWaitChunk_busy, evf::FastMonState::inWaitChunk_fileLimit, evf::FastMonState::inWaitChunk_lockPolling, evf::FastMonState::inWaitChunk_lockPollingCopying, evf::FastMonState::inWaitChunk_newFile, evf::FastMonState::inWaitChunk_newFileWaitChunk, evf::FastMonState::inWaitChunk_newFileWaitChunkCopying, evf::FastMonState::inWaitChunk_newFileWaitThread, evf::FastMonState::inWaitChunk_newFileWaitThreadCopying, evf::FastMonState::inWaitChunk_noFile, evf::FastMonState::inWaitChunk_runEnd, evf::FastMonState::inWaitChunk_waitFreeChunk, evf::FastMonState::inWaitChunk_waitFreeChunkCopying, evf::FastMonState::inWaitChunk_waitFreeThread, evf::FastMonState::inWaitChunk_waitFreeThreadCopying, evf::FastMonState::inWaitInput, evf::FastMonState::inWaitInput_busy, evf::FastMonState::inWaitInput_fileLimit, evf::FastMonState::inWaitInput_lockPolling, evf::FastMonState::inWaitInput_lockPollingCopying, evf::FastMonState::inWaitInput_newFile, evf::FastMonState::inWaitInput_newFileWaitChunk, evf::FastMonState::inWaitInput_newFileWaitChunkCopying, evf::FastMonState::inWaitInput_newFileWaitThread, evf::FastMonState::inWaitInput_newFileWaitThreadCopying, evf::FastMonState::inWaitInput_noFile, evf::FastMonState::inWaitInput_runEnd, evf::FastMonState::inWaitInput_waitFreeChunk, evf::FastMonState::inWaitInput_waitFreeChunkCopying, evf::FastMonState::inWaitInput_waitFreeThread, evf::FastMonState::inWaitInput_waitFreeThreadCopying, isInitTransition_, lockStatsDuringLumi_, eostools::ls(), microstate_, microstateAcqFlag_, nMonThreads_, nStreams_, nThreads_, tmicrostate_, and tmicrostateAcqFlag_.

Referenced by preGlobalEndLumi(), and snapshotRunner().

                                                                                      {
    // update macrostate
    fmt_->m_data.fastMacrostateJ_ = fmt_->m_data.macrostate_;

    std::vector<const void*> microstateCopy(microstate_.begin(), microstate_.end());
    std::vector<const void*> tmicrostateCopy(tmicrostate_.begin(), tmicrostate_.end());
    std::vector<unsigned char> microstateAcqCopy(microstateAcqFlag_.begin(), microstateAcqFlag_.end());
    std::vector<unsigned char> tmicrostateAcqCopy(tmicrostateAcqFlag_.begin(), tmicrostateAcqFlag_.end());

    if (!isInitTransition_) {
      auto itd = avgLeadTime_.find(ls);
      if (itd != avgLeadTime_.end())
        fmt_->m_data.fastAvgLeadTimeJ_ = itd->second;
      else
        fmt_->m_data.fastAvgLeadTimeJ_ = 0.;

      auto iti = filesProcessedDuringLumi_.find(ls);
      if (iti != filesProcessedDuringLumi_.end())
        fmt_->m_data.fastFilesProcessedJ_ = iti->second;
      else
        fmt_->m_data.fastFilesProcessedJ_ = 0;

      auto itrd = lockStatsDuringLumi_.find(ls);
      if (itrd != lockStatsDuringLumi_.end()) {
        fmt_->m_data.fastLockWaitJ_ = itrd->second.first;
        fmt_->m_data.fastLockCountJ_ = itrd->second.second;
      } else {
        fmt_->m_data.fastLockWaitJ_ = 0.;
        fmt_->m_data.fastLockCountJ_ = 0.;
      }
    }

    for (unsigned int i = 0; i < nThreads_; i++) {
      if (tmicrostateCopy[i] == getmIdle() && ct_->isThreadActive(i)) {
        //overhead if thread is running
        tmicrostateCopy[i] = getmFwk();
      }
      if (tmicrostateAcqCopy[i])
        fmt_->m_data.tmicrostateEncoded_[i] =
            fmt_->m_data.microstateBins_ + fmt_->m_data.encModule_.encode(tmicrostateCopy[i]);
      else
        fmt_->m_data.tmicrostateEncoded_[i] = fmt_->m_data.encModule_.encode(tmicrostateCopy[i]);
    }

    for (unsigned int i = 0; i < nStreams_; i++) {
      if (microstateAcqCopy[i])
        fmt_->m_data.microstateEncoded_[i] =
            fmt_->m_data.microstateBins_ + fmt_->m_data.encModule_.encode(microstateCopy[i]);
      else
        fmt_->m_data.microstateEncoded_[i] = fmt_->m_data.encModule_.encode(microstateCopy[i]);
    }

    bool inputStatePerThread = false;

    if (inputState_ == FastMonState::inWaitInput) {
      switch (inputSupervisorState_) {
        case FastMonState::inSupFileLimit:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_fileLimit;
          break;
        case FastMonState::inSupWaitFreeChunk:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeChunk;
          break;
        case FastMonState::inSupWaitFreeChunkCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeChunkCopying;
          break;
        case FastMonState::inSupWaitFreeThread:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeThread;
          break;
        case FastMonState::inSupWaitFreeThreadCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeThreadCopying;
          break;
        case FastMonState::inSupBusy:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_busy;
          break;
        case FastMonState::inSupLockPolling:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_lockPolling;
          break;
        case FastMonState::inSupLockPollingCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_lockPollingCopying;
          break;
        case FastMonState::inRunEnd:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_runEnd;
          break;
        case FastMonState::inSupNoFile:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_noFile;
          break;
        case FastMonState::inSupNewFile:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFile;
          break;
        case FastMonState::inSupNewFileWaitThreadCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitThreadCopying;
          break;
        case FastMonState::inSupNewFileWaitThread:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitThread;
          break;
        case FastMonState::inSupNewFileWaitChunkCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitChunkCopying;
          break;
        case FastMonState::inSupNewFileWaitChunk:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitChunk;
          break;
        default:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitInput;
      }
    } else if (inputState_ == FastMonState::inWaitChunk) {
      switch (inputSupervisorState_) {
        case FastMonState::inSupFileLimit:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_fileLimit;
          break;
        case FastMonState::inSupWaitFreeChunk:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeChunk;
          break;
        case FastMonState::inSupWaitFreeChunkCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeChunkCopying;
          break;
        case FastMonState::inSupWaitFreeThread:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeThread;
          break;
        case FastMonState::inSupWaitFreeThreadCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeThreadCopying;
          break;
        case FastMonState::inSupBusy:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_busy;
          break;
        case FastMonState::inSupLockPolling:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_lockPolling;
          break;
        case FastMonState::inSupLockPollingCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_lockPollingCopying;
          break;
        case FastMonState::inRunEnd:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_runEnd;
          break;
        case FastMonState::inSupNoFile:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_noFile;
          break;
        case FastMonState::inSupNewFile:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFile;
          break;
        case FastMonState::inSupNewFileWaitThreadCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitThreadCopying;
          break;
        case FastMonState::inSupNewFileWaitThread:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitThread;
          break;
        case FastMonState::inSupNewFileWaitChunkCopying:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitChunkCopying;
          break;
        case FastMonState::inSupNewFileWaitChunk:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitChunk;
          break;
        default:
          fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk;
      }
    } else if (inputState_ == FastMonState::inNoRequest) {
      inputStatePerThread = true;
      for (unsigned int i = 0; i < nMonThreads_; i++) {
        if (i >= nStreams_)
          fmt_->m_data.inputState_[i] = FastMonState::inIgnore;
        else if (microstateCopy[i] == getmIdle())
          fmt_->m_data.inputState_[i] = FastMonState::inNoRequestWithIdleThreads;
        else if (microstateCopy[i] == getmEoL() || microstateCopy[i] == getmFwkEoL())
          fmt_->m_data.inputState_[i] = FastMonState::inNoRequestWithEoLThreads;
        else
          fmt_->m_data.inputState_[i] = FastMonState::inNoRequest;
      }
    } else if (inputState_ == FastMonState::inNewLumi) {
      inputStatePerThread = true;
      for (unsigned int i = 0; i < nMonThreads_; i++) {
        if (i >= nStreams_)
          fmt_->m_data.inputState_[i] = FastMonState::inIgnore;
        else if (microstateCopy[i] == getmEoL() || microstateCopy[i] == getmFwkEoL())
          fmt_->m_data.inputState_[i] = FastMonState::inNewLumi;
      }
    } else if (inputSupervisorState_ == FastMonState::inSupThrottled) {
      //apply directly throttled state from supervisor
      fmt_->m_data.inputState_[0] = inputSupervisorState_;
    } else
      fmt_->m_data.inputState_[0] = inputState_;

    //this is same for all streams
    if (!inputStatePerThread)
      for (unsigned int i = 1; i < nMonThreads_; i++)
        fmt_->m_data.inputState_[i] = fmt_->m_data.inputState_[0];

    if (isGlobalEOL) {  //only update global variables
      fmt_->jsonMonitor_->snapGlobal(ls);
    } else
      fmt_->jsonMonitor_->snap(ls);
  }

exceptionDetected()

bool evf::FastMonitoringService::exceptionDetected

(

)

const

Definition at line 457 of file FastMonitoringService.cc.

References has_data_exception_, and has_source_exception_.

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

                                                      {
    return has_source_exception_.load() || has_data_exception_.load();
  }

fillDescriptions()

void evf::FastMonitoringService::fillDescriptions ( edm::ConfigurationDescriptions &  descriptions )

static

Definition at line 258 of file FastMonitoringService.cc.

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

                                                                                         {
    edm::ParameterSetDescription desc;
    desc.setComment("Service for File-based DAQ monitoring and event accounting");
    desc.addUntracked<bool>("tbbMonitoringMode", true)
        ->setComment("Monitor individual module processing per TBB thread instead of stream");
    desc.addUntracked<bool>("tbbConcurrencyTracker", true)
        ->setComment("Monitor TBB thread activity to flag microstate as real idle or overhead/other");
    desc.addUntracked<int>("sleepTime", 1)->setComment("Sleep time of the monitoring thread");
    desc.addUntracked<unsigned int>("fastMonIntervals", 2)
        ->setComment("Modulo of sleepTime intervals on which fastmon file is written out");
    desc.addUntracked<bool>("filePerFwkStream", true)  //obsolete
        ->setComment("Switches on monitoring output per framework stream");
    desc.addUntracked<bool>("verbose", false)->setComment("Set to use LogInfo messages from the monitoring thread");
    desc.setAllowAnything();
    descriptions.add("FastMonitoringService", desc);
  }

getAbortFlagForLumi()

bool evf::FastMonitoringService::getAbortFlagForLumi

(

unsigned int

lumi

)

Definition at line 795 of file FastMonitoringService.cc.

References Exception, fmt_, ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, CommonMethods::lock(), BXlumiParameters_cfi::lumi, and processedEventsPerLumi_.

Referenced by shouldWriteFiles().

                                                                   {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);

    auto it = processedEventsPerLumi_.find(lumi);
    if (it != processedEventsPerLumi_.end()) {
      unsigned int abortFlag = it->second.second;
      return abortFlag;
    } else {
      throw cms::Exception("FastMonitoringService")
          << "output module wants already deleted (or never reported by SOURCE) lumisection status for LUMI -: "
          << lumi;
      return false;
    }
  }

getEventsProcessedForLumi()

unsigned int evf::FastMonitoringService::getEventsProcessedForLumi	(	unsigned int	lumi,
		bool *	abortFlag = nullptr
	)

Definition at line 777 of file FastMonitoringService.cc.

References Exception, fmt_, ALPAKA_ACCELERATOR_NAMESPACE::vertexFinder::it, CommonMethods::lock(), BXlumiParameters_cfi::lumi, ValidateTausOnZEEFastSim_cff::proc, and processedEventsPerLumi_.

Referenced by dqm::DQMFileSaverPB::fillJson(), dqmfilesaver::fillJson(), evf::GlobalEvFOutputModule::globalEndLuminosityBlock(), dqm::DQMFileSaverPB::saveLumi(), and shouldWriteFiles().

                                                                                                  {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);

    auto it = processedEventsPerLumi_.find(lumi);
    if (it != processedEventsPerLumi_.end()) {
      unsigned int proc = it->second.first;
      if (abortFlag)
        *abortFlag = it->second.second;
      return proc;
    } else {
      throw cms::Exception("FastMonitoringService")
          << "output module wants already deleted (or never reported by SOURCE) lumisection event count for LUMI -: "
          << lumi;
      return 0;
    }
  }

getRunDirName()

std::string evf::FastMonitoringService::getRunDirName ( ) const

inline

Definition at line 226 of file FastMonitoringService.h.

References runDirectory_.

{ return runDirectory_.stem().string(); }

getSID() [1/2]

static unsigned int evf::FastMonitoringService::getSID ( edm::StreamContext const &  sc )

inlinestaticprivate

Definition at line 241 of file FastMonitoringService.h.

References edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), preModuleEvent(), preModuleEventAcquire(), and preSourceEvent().

{ return sc.streamID().value(); }

getSID() [2/2]

static unsigned int evf::FastMonitoringService::getSID ( edm::StreamID const &  sid )

inlinestaticprivate

Definition at line 243 of file FastMonitoringService.h.

References edm::StreamID::value().

{ return sid.value(); }

getTID()

static unsigned int evf::FastMonitoringService::getTID ( )

inlinestatic

Definition at line 234 of file FastMonitoringService.h.

Referenced by postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), preModuleEvent(), preModuleEventAcquire(), and preSourceEvent().

{ return tbb::this_task_arena::current_thread_index(); }

isExceptionOnData()

bool evf::FastMonitoringService::isExceptionOnData

(

unsigned int

ls

)

Definition at line 461 of file FastMonitoringService.cc.

References exceptionInLS_, fmt_, has_data_exception_, has_source_exception_, CommonMethods::lock(), and eostools::ls().

Referenced by evf::EvFDaqDirector::preGlobalEndLumi(), FedRawDataInputSource::read(), DAQSource::read(), DAQSource::~DAQSource(), and FedRawDataInputSource::~FedRawDataInputSource().

                                                               {
    if (!has_data_exception_.load())
      return false;
    if (has_source_exception_.load())
      return true;
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    for (auto ex : exceptionInLS_) {
      if (ls == ex)
        return true;
    }
    return false;
  }

jobFailure()

void evf::FastMonitoringService::jobFailure

(

)

Definition at line 474 of file FastMonitoringService.cc.

References fmt_, and evf::FastMonState::sError.

Referenced by FastMonitoringService().

{ fmt_->m_data.macrostate_ = FastMonState::sError; }

makeInputLegendaJson()

std::string evf::FastMonitoringService::makeInputLegendaJson

(

)

Definition at line 296 of file FastMonitoringService.cc.

References jsoncollector::Json::Value::append(), jsoncollector::Json::arrayValue, mps_fire::i, evf::FastMonState::inCOUNT, inputStateNames, and convertToRaw::writer.

Referenced by postBeginJob().

                                                        {
    Json::Value legendaVector(Json::arrayValue);
    for (int i = 0; i < FastMonState::inCOUNT; i++)
      legendaVector.append(Json::Value(inputStateNames[i]));
    Json::Value moduleLegend;
    moduleLegend["names"] = legendaVector;
    Json::StyledWriter writer;
    return writer.write(moduleLegend);
  }

makeModuleLegendaJson()

std::string evf::FastMonitoringService::makeModuleLegendaJson

(

)

Definition at line 275 of file FastMonitoringService.cc.

References jsoncollector::Json::Value::append(), jsoncollector::Json::arrayValue, fmt_, mps_fire::i, nOutputModules_, evf::nReservedModules, evf::nSpecialModules, and convertToRaw::writer.

Referenced by postBeginJob().

                                                         {
    Json::Value legendaVector(Json::arrayValue);
    for (int i = 0; i < fmt_->m_data.encModule_.current_; i++)
      legendaVector.append(
          Json::Value((static_cast<const edm::ModuleDescription*>(fmt_->m_data.encModule_.decode(i)))->moduleLabel()));
    //duplicate modules adding a list for acquire states (not all modules actually have it)
    for (int i = 0; i < fmt_->m_data.encModule_.current_; i++)
      legendaVector.append(Json::Value(
          (static_cast<const edm::ModuleDescription*>(fmt_->m_data.encModule_.decode(i)))->moduleLabel() + "__ACQ"));
    Json::Value valReserved(nReservedModules);
    Json::Value valSpecial(nSpecialModules);
    Json::Value valOutputModules(nOutputModules_);
    Json::Value moduleLegend;
    moduleLegend["names"] = legendaVector;
    moduleLegend["reserved"] = valReserved;
    moduleLegend["special"] = valSpecial;
    moduleLegend["output"] = valOutputModules;
    Json::StyledWriter writer;
    return writer.write(moduleLegend);
  }

postBeginJob()

void evf::FastMonitoringService::postBeginJob

(

)

Definition at line 491 of file FastMonitoringService.cc.

References fmt_, inputLegendFileJson_, CommonMethods::lock(), makeInputLegendaJson(), makeModuleLegendaJson(), moduleLegendFileJson_, evf::FastMonState::sJobReady, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FastMonitoringService().

                                           {
    std::string&& moduleLegStrJson = makeModuleLegendaJson();
    FileIO::writeStringToFile(moduleLegendFileJson_, moduleLegStrJson);

    std::string inputLegendStrJson = makeInputLegendaJson();
    FileIO::writeStringToFile(inputLegendFileJson_, inputLegendStrJson);

    fmt_->m_data.macrostate_ = FastMonState::sJobReady;

    //update number of entries in module histogram
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    //double the size to add post-acquire states
    fmt_->m_data.microstateBins_ = fmt_->m_data.encModule_.vecsize() * 2;
  }

postEndJob()

void evf::FastMonitoringService::postEndJob

(

)

Definition at line 506 of file FastMonitoringService.cc.

References fmt_, and evf::FastMonState::sJobEnded.

Referenced by FastMonitoringService().

                                         {
    fmt_->m_data.macrostate_ = FastMonState::sJobEnded;
    fmt_->stop();
  }

postEvent()

void evf::FastMonitoringService::postEvent

(

edm::StreamContext const &

sc

)

Definition at line 641 of file FastMonitoringService.cc.

References fmt_, evf::getmIdle(), microstate_, edm::StreamContext::streamID(), totalEventsProcessed_, and edm::StreamID::value().

Referenced by FastMonitoringService().

                                                                  {
    (*(fmt_->m_data.processed_[sc.streamID().value()]))++;
    //fast path counter (events accumulated in a run)
    unsigned long res = totalEventsProcessed_.fetch_add(1, std::memory_order_relaxed);
    fmt_->m_data.fastPathProcessedJ_ = res + 1;

    microstate_[sc.streamID().value()] = getmIdle();
  }

postGlobalBeginRun()

void evf::FastMonitoringService::postGlobalBeginRun

(

edm::GlobalContext const &

gc

)

Definition at line 511 of file FastMonitoringService.cc.

References fmt_, isInitTransition_, and evf::FastMonState::sRunning.

                                                                           {
    fmt_->m_data.macrostate_ = FastMonState::sRunning;
    isInitTransition_ = false;
  }

postGlobalEndLumi()

void evf::FastMonitoringService::postGlobalEndLumi

(

edm::GlobalContext const &

gc

)

Definition at line 598 of file FastMonitoringService.cc.

References avgLeadTime_, filesProcessedDuringLumi_, fmt_, CommonMethods::lock(), lockStatsDuringLumi_, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), and processedEventsPerLumi_.

Referenced by FastMonitoringService().

                                                                          {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
    //LS monitoring snapshot with input source data has been taken in previous callback
    avgLeadTime_.erase(lumi);
    filesProcessedDuringLumi_.erase(lumi);
    lockStatsDuringLumi_.erase(lumi);

    //output module already used this in end lumi (this could be migrated to EvFDaqDirector as it is essential for FFF bookkeeping)
    processedEventsPerLumi_.erase(lumi);
  }

postModuleEvent()

void evf::FastMonitoringService::postModuleEvent	(	edm::StreamContext const &	sc,
		edm::ModuleCallingContext const &	mcc
	)

Definition at line 706 of file FastMonitoringService.cc.

References evf::getmFwkOvhMod(), evf::getmIdle(), getSID(), getTID(), microstate_, nThreads_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

                                                                                                            {
    microstate_[getSID(sc)] = getmFwkOvhMod();
    if (!tbbMonitoringMode_)
      return;
    auto tid = getTID();
    if (tid >= nThreads_)
      return;
    tmicrostate_[tid] = getmIdle();
  }

postModuleEventAcquire()

void evf::FastMonitoringService::postModuleEventAcquire	(	edm::StreamContext const &	sc,
		edm::ModuleCallingContext const &	mcc
	)

Definition at line 683 of file FastMonitoringService.cc.

References evf::getmFwkOvhMod(), evf::getmIdle(), getSID(), getTID(), microstate_, microstateAcqFlag_, nThreads_, tbbMonitoringMode_, tmicrostate_, and tmicrostateAcqFlag_.

Referenced by FastMonitoringService().

                                                                                         {
    microstate_[getSID(sc)] = getmFwkOvhMod();
    microstateAcqFlag_[getSID(sc)] = 0;
    if (!tbbMonitoringMode_)
      return;
    auto tid = getTID();
    if (tid >= nThreads_)
      return;
    tmicrostate_[tid] = getmIdle();
    tmicrostateAcqFlag_[tid] = 0;
  }

postSourceEvent()

void evf::FastMonitoringService::postSourceEvent

(

edm::StreamID

sid

)

Definition at line 660 of file FastMonitoringService.cc.

References evf::getmFwkOvhSrc(), evf::getmIdle(), getSID(), getTID(), microstate_, nThreads_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

                                                             {
    microstate_[getSID(sid)] = getmFwkOvhSrc();
    if (!tbbMonitoringMode_)
      return;
    auto tid = getTID();
    if (tid >= nThreads_)
      return;
    tmicrostate_[tid] = getmIdle();
  }

postStreamBeginLumi()

void evf::FastMonitoringService::postStreamBeginLumi

(

edm::StreamContext const &

sc

)

Definition at line 620 of file FastMonitoringService.cc.

References evf::getmIdle(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

                                                                            {
    microstate_[sc.streamID().value()] = getmIdle();
  }

postStreamEndLumi()

void evf::FastMonitoringService::postStreamEndLumi

(

edm::StreamContext const &

sc

)

Definition at line 633 of file FastMonitoringService.cc.

References evf::getmFwkEoL(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

                                                                          {
    microstate_[sc.streamID().value()] = getmFwkEoL();
  }

preallocate()

void evf::FastMonitoringService::preallocate

(

edm::service::SystemBounds const &

bounds

)

Definition at line 306 of file FastMonitoringService.cc.

References ct_, SiStripPI::max, nMonThreads_, nStreams_, and nThreads_.

Referenced by FastMonitoringService().

                                                                              {
    nStreams_ = bounds.maxNumberOfStreams();
    nThreads_ = bounds.maxNumberOfThreads();
    //this should already be >=1
    if (nStreams_ == 0)
      nStreams_ = 1;
    if (nThreads_ == 0)
      nThreads_ = 1;
    nMonThreads_ = std::max(nThreads_, nStreams_);
    ct_ = std::make_unique<ConcurrencyTracker>(nThreads_);
    //start concurrency tracking
  }

preBeginJob()

void evf::FastMonitoringService::preBeginJob	(	edm::PathsAndConsumesOfModulesBase const &	,
		edm::ProcessContext const &	pc
	)

Definition at line 319 of file FastMonitoringService.cc.

References ct_, Exception, fastMicrostateDefPath_, fastName_, fastPath_, fmt_, evf::getmInvalid(), mps_fire::i, evf::FastMonState::inCOUNT, inputLegendFileJson_, isInitTransition_, lastGlobalLumi_, LogDebug, lumiFromSource_, evf::FastMonState::mCOUNT, evf::FastMonState::MCOUNT, microstate_, microstateAcqFlag_, microstateDefPath_, moduleLegendFile_, moduleLegendFileJson_, monInit_, nMonThreads_, nStreams_, nThreads_, Utilities::operator, castor_dqm_sourceclient_file_cfg::path, runDirectory_, evf::FastMonState::sInit, sleepTime_, snapshotRunner(), specialMicroStateNames, streamCounterUpdating_, tbbConcurrencyTracker_, tmicrostate_, tmicrostateAcqFlag_, and workingDirectory_.

Referenced by FastMonitoringService().

                                                                                                              {
    // FIND RUN DIRECTORY
    // The run dir should be set via the configuration of EvFDaqDirector
    if (tbbConcurrencyTracker_)
      ct_->activate();

    if (edm::Service<evf::EvFDaqDirector>().operator->() == nullptr) {
      throw cms::Exception("FastMonitoringService") << "EvFDaqDirector is not present";
    }
    std::filesystem::path runDirectory{edm::Service<evf::EvFDaqDirector>()->baseRunDir()};
    workingDirectory_ = runDirectory_ = runDirectory;
    workingDirectory_ /= "mon";

    if (!std::filesystem::is_directory(workingDirectory_)) {
      LogDebug("FastMonitoringService") << "<MON> DIR NOT FOUND! Trying to create -: " << workingDirectory_.string();
      std::filesystem::create_directories(workingDirectory_);
      if (!std::filesystem::is_directory(workingDirectory_))
        edm::LogWarning("FastMonitoringService") << "Unable to create <MON> DIR -: " << workingDirectory_.string()
                                                 << ". No monitoring data will be written.";
    }

    std::ostringstream fastFileName;

    fastFileName << fastName_ << "_pid" << std::setfill('0') << std::setw(5) << getpid() << ".fast";
    std::filesystem::path fast = workingDirectory_;
    fast /= fastFileName.str();
    fastPath_ = fast.string();

    std::ostringstream moduleLegFile;
    std::ostringstream moduleLegFileJson;
    moduleLegFile << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
    moduleLegFileJson << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
    moduleLegendFile_ = (workingDirectory_ / moduleLegFile.str()).string();
    moduleLegendFileJson_ = (workingDirectory_ / moduleLegFileJson.str()).string();

    std::ostringstream inputLegFileJson;
    inputLegFileJson << "inputlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
    inputLegendFileJson_ = (workingDirectory_ / inputLegFileJson.str()).string();

    LogDebug("FastMonitoringService") << "Initializing FastMonitor with microstate def path -: " << microstateDefPath_;

    /*
     * initialize the fast monitor with:
     * vector of pointers to monitorable parameters
     * path to definition
     *
     */

    fmt_->m_data.macrostate_ = FastMonState::sInit;

    for (unsigned int i = 0; i < (FastMonState::mCOUNT); i++)
      fmt_->m_data.encModule_.updateReserved(static_cast<const void*>(specialMicroStateNames + i));
    fmt_->m_data.encModule_.completeReservedWithDummies();

    for (unsigned int i = 0; i < nMonThreads_; i++) {
      microstate_.emplace_back(getmInvalid());
      microstateAcqFlag_.push_back(0);
      tmicrostate_.emplace_back(getmInvalid());
      tmicrostateAcqFlag_.push_back(0);

      //for synchronization
      streamCounterUpdating_.push_back(new std::atomic<bool>(false));
    }

    //initial size until we detect number of bins
    fmt_->m_data.macrostateBins_ = FastMonState::MCOUNT;
    fmt_->m_data.microstateBins_ = 0;
    fmt_->m_data.inputstateBins_ = FastMonState::inCOUNT;

    lastGlobalLumi_ = 0;
    isInitTransition_ = true;
    lumiFromSource_ = 0;

    //startup monitoring
    fmt_->resetFastMonitor(microstateDefPath_, fastMicrostateDefPath_);
    fmt_->jsonMonitor_->setNStreams(nMonThreads_);
    fmt_->m_data.registerVariables(fmt_->jsonMonitor_.get(), nMonThreads_, nStreams_, nThreads_);
    monInit_.store(false, std::memory_order_release);
    if (sleepTime_ > 0)
      fmt_->start(&FastMonitoringService::snapshotRunner, this);
  }

preEvent()

void evf::FastMonitoringService::preEvent

(

edm::StreamContext const &

sc

)

Definition at line 637 of file FastMonitoringService.cc.

References evf::getmEvent(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

                                                                 {
    microstate_[sc.streamID().value()] = getmEvent();
  }

preGlobalBeginLumi()

void evf::FastMonitoringService::preGlobalBeginLumi

(

edm::GlobalContext const &

gc

)

Definition at line 516 of file FastMonitoringService.cc.

References fmt_, evf::getmIdle(), evf::getmInvalid(), mps_fire::i, lastGlobalLumi_, CommonMethods::lock(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, nThreads_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

                                                                           {
    timeval lumiStartTime;
    gettimeofday(&lumiStartTime, nullptr);
    unsigned int newLumi = gc.luminosityBlockID().luminosityBlock();
    lastGlobalLumi_ = newLumi;

    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    lumiStartTime_[newLumi] = lumiStartTime;
    //reset all states to idle
    if (tbbMonitoringMode_)
      for (unsigned i = 0; i < nThreads_; i++)
        if (tmicrostate_[i] == getmInvalid())
          tmicrostate_[i] = getmIdle();
  }

preGlobalEarlyTermination()

void evf::FastMonitoringService::preGlobalEarlyTermination	(	edm::GlobalContext const &	gc,
		edm::TerminationOrigin	to
	)

Definition at line 417 of file FastMonitoringService.cc.

References visDQMUpload::context, edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, exceptionInLS_, edm::ExternalSignal, fmt_, has_data_exception_, CommonMethods::lock(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FastMonitoringService().

                                                                                                           {
    std::string context;
    if (to == edm::TerminationOrigin::ExceptionFromThisContext)
      context = " FromThisContext ";
    if (to == edm::TerminationOrigin::ExceptionFromAnotherContext)
      context = " FromAnotherContext";
    if (to == edm::TerminationOrigin::ExternalSignal)
      context = " FromExternalSignal";
    edm::LogWarning("FastMonitoringService")
        << " GLOBAL "
        << "earlyTermination -: LS:" << gc.luminosityBlockID().luminosityBlock() << " " << context;
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    exceptionInLS_.push_back(gc.luminosityBlockID().luminosityBlock());
    has_data_exception_.store(true);
  }

preGlobalEndLumi()

void evf::FastMonitoringService::preGlobalEndLumi

(

edm::GlobalContext const &

gc

)

Definition at line 531 of file FastMonitoringService.cc.

References accuSize_, daqInputSource_, doSnapshot(), Exception, exception_detected_, exceptionInLS_, fmt_, DAQSource::getEventReport(), FedRawDataInputSource::getEventReport(), inputSource_, CommonMethods::lock(), LogDebug, BXlumiParameters_cfi::lumi, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, processedEventsPerLumi_, edm::shutdown_flag, throughputFactor(), and jsoncollector::IntJ::value().

Referenced by FastMonitoringService().

                                                                         {
    unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
    LogDebug("FastMonitoringService") << "Lumi ended. Writing JSON information. LUMI -: " << lumi;
    timeval lumiStopTime;
    gettimeofday(&lumiStopTime, nullptr);

    std::lock_guard<std::mutex> lock(fmt_->monlock_);

    // Compute throughput
    timeval stt = lumiStartTime_[lumi];
    lumiStartTime_.erase(lumi);
    unsigned long usecondsForLumi = (lumiStopTime.tv_sec - stt.tv_sec) * 1000000 + (lumiStopTime.tv_usec - stt.tv_usec);
    unsigned long accuSize = accuSize_.find(lumi) == accuSize_.end() ? 0 : accuSize_[lumi];
    accuSize_.erase(lumi);
    double throughput = throughputFactor() * double(accuSize) / double(usecondsForLumi);
    //store to registered variable
    fmt_->m_data.fastThroughputJ_.value() = throughput;

    //update
    doSnapshot(lumi, true);

    //retrieve one result we need (todo: sanity check if it's found)
    IntJ* lumiProcessedJptr = dynamic_cast<IntJ*>(fmt_->jsonMonitor_->getMergedIntJForLumi("Processed", lumi));
    if (!lumiProcessedJptr)
      throw cms::Exception("FastMonitoringService") << "Internal error: got null pointer from FastMonitor";
    processedEventsPerLumi_[lumi] = std::pair<unsigned int, bool>(lumiProcessedJptr->value(), false);

    //checking if exception has been thrown (in case of Global/Stream early termination, for this LS)
    bool exception_detected = exception_detected_;
    for (auto ex : exceptionInLS_)
      if (lumi == ex)
        exception_detected = true;

    if (edm::shutdown_flag || exception_detected) {
      edm::LogInfo("FastMonitoringService")
          << "Run interrupted. Skip writing EoL information -: " << processedEventsPerLumi_[lumi].first
          << " events were processed in LUMI " << lumi;
      //this will prevent output modules from producing json file for possibly incomplete lumi
      processedEventsPerLumi_[lumi].first = 0;
      processedEventsPerLumi_[lumi].second = true;
      //disable this exception, so service can be used standalone (will be thrown if output module asks for this information)
      //throw cms::Exception("FastMonitoringService") << "SOURCE did not send update for lumi block. LUMI -:" << lumi;
      return;
    }

    if (inputSource_ || daqInputSource_) {
      auto sourceReport =
          inputSource_ ? inputSource_->getEventReport(lumi, true) : daqInputSource_->getEventReport(lumi, true);
      if (sourceReport.first) {
        if (sourceReport.second != processedEventsPerLumi_[lumi].first) {
          throw cms::Exception("FastMonitoringService") << "MISMATCH with SOURCE update. LUMI -: " << lumi
                                                        << ", events(processed):" << processedEventsPerLumi_[lumi].first
                                                        << " events(source):" << sourceReport.second;
        }
      }
    }

    edm::LogInfo("FastMonitoringService")
        << "Statistics for lumisection -: lumi = " << lumi << " events = " << lumiProcessedJptr->value()
        << " time = " << usecondsForLumi / 1000000 << " size = " << accuSize << " thr = " << throughput;
    delete lumiProcessedJptr;

    //full global and stream merge (will be used by output modules), output from this service is deprecated
    fmt_->jsonMonitor_->outputFullJSON("dummy", lumi, false);
    fmt_->jsonMonitor_->discardCollected(lumi);  //we don't do further updates for this lumi
  }

preModuleBeginJob()

void evf::FastMonitoringService::preModuleBeginJob

(

edm::ModuleDescription const &

desc

)

Definition at line 477 of file FastMonitoringService.cc.

References submitPVResolutionJobs::desc, fmt_, CommonMethods::lock(), and nOutputModules_.

Referenced by FastMonitoringService().

                                                                                {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    //std::cout << " Pre module Begin Job module: " << desc.moduleName() << std::endl;

    //build a map of modules keyed by their module description address
    //here we need to treat output modules in a special way so they can be easily singled out
    if (desc.moduleName() == "Stream" || desc.moduleName() == "GlobalEvFOutputModule" ||
        desc.moduleName() == "EventStreamFileWriter" || desc.moduleName() == "PoolOutputModule") {
      fmt_->m_data.encModule_.updateReserved((void*)&desc);
      nOutputModules_++;
    } else
      fmt_->m_data.encModule_.update((void*)&desc);
  }

preModuleEvent()

void evf::FastMonitoringService::preModuleEvent	(	edm::StreamContext const &	sc,
		edm::ModuleCallingContext const &	mcc
	)

Definition at line 696 of file FastMonitoringService.cc.

References getSID(), getTID(), microstate_, edm::ModuleCallingContext::moduleDescription(), nThreads_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

                                                                                                           {
    microstate_[getSID(sc)] = (void*)(mcc.moduleDescription());
    if (!tbbMonitoringMode_)
      return;
    auto tid = getTID();
    if (tid >= nThreads_)
      return;
    tmicrostate_[tid] = (void*)(mcc.moduleDescription());
  }

preModuleEventAcquire()

void evf::FastMonitoringService::preModuleEventAcquire	(	edm::StreamContext const &	sc,
		edm::ModuleCallingContext const &	mcc
	)

Definition at line 670 of file FastMonitoringService.cc.

References getSID(), getTID(), microstate_, microstateAcqFlag_, edm::ModuleCallingContext::moduleDescription(), nThreads_, tbbMonitoringMode_, tmicrostate_, and tmicrostateAcqFlag_.

Referenced by FastMonitoringService().

                                                                                        {
    microstate_[getSID(sc)] = (void*)(mcc.moduleDescription());
    microstateAcqFlag_[getSID(sc)] = 1;
    if (!tbbMonitoringMode_)
      return;
    auto tid = getTID();
    if (tid >= nThreads_)
      return;
    tmicrostate_[tid] = (void*)(mcc.moduleDescription());
    tmicrostateAcqFlag_[tid] = 1;
  }

prePathEvent()

void evf::FastMonitoringService::prePathEvent	(	edm::StreamContext const &	,
		edm::PathContext const &
	)

preSourceEarlyTermination()

void evf::FastMonitoringService::preSourceEarlyTermination

(

edm::TerminationOrigin

to

)

Definition at line 433 of file FastMonitoringService.cc.

References visDQMUpload::context, exception_detected_, edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, edm::ExternalSignal, fmt_, has_data_exception_, has_source_exception_, CommonMethods::lock(), and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FastMonitoringService().

                                                                               {
    std::string context;
    if (to == edm::TerminationOrigin::ExceptionFromThisContext)
      context = " FromThisContext ";
    if (to == edm::TerminationOrigin::ExceptionFromAnotherContext)
      context = " FromAnotherContext";
    if (to == edm::TerminationOrigin::ExternalSignal)
      context = " FromExternalSignal";
    edm::LogWarning("FastMonitoringService") << " SOURCE "
                                             << "earlyTermination -: " << context;
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    exception_detected_ = true;
    has_source_exception_.store(true);
    has_data_exception_.store(true);
  }

preSourceEvent()

void evf::FastMonitoringService::preSourceEvent

(

edm::StreamID

sid

)

Definition at line 650 of file FastMonitoringService.cc.

References evf::getmInput(), getSID(), getTID(), microstate_, nThreads_, tbbMonitoringMode_, and tmicrostate_.

Referenced by FastMonitoringService().

                                                            {
    microstate_[getSID(sid)] = getmInput();
    if (!tbbMonitoringMode_)
      return;
    auto tid = getTID();
    if (tid >= nThreads_)
      return;
    tmicrostate_[tid] = getmInput();
  }

preStreamBeginLumi()

void evf::FastMonitoringService::preStreamBeginLumi

(

edm::StreamContext const &

sc

)

Definition at line 610 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), fmt_, evf::getmBoL(), CommonMethods::lock(), edm::EventID::luminosityBlock(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

                                                                           {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    fmt_->m_data.streamLumi_[sc.streamID().value()] = sc.eventID().luminosityBlock();

    //reset collected values for this stream
    *(fmt_->m_data.processed_[sc.streamID().value()]) = 0;

    microstate_[sc.streamID().value()] = getmBoL();
  }

preStreamEarlyTermination()

void evf::FastMonitoringService::preStreamEarlyTermination	(	edm::StreamContext const &	sc,
		edm::TerminationOrigin	to
	)

Definition at line 401 of file FastMonitoringService.cc.

References visDQMUpload::context, edm::StreamContext::eventID(), edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, exceptionInLS_, edm::ExternalSignal, fmt_, has_data_exception_, CommonMethods::lock(), edm::EventID::luminosityBlock(), edm::StreamContext::streamID(), AlCaHLTBitMon_QueryRunRegistry::string, and edm::StreamID::value().

Referenced by FastMonitoringService().

                                                                                                           {
    std::string context;
    if (to == edm::TerminationOrigin::ExceptionFromThisContext)
      context = " FromThisContext ";
    if (to == edm::TerminationOrigin::ExceptionFromAnotherContext)
      context = " FromAnotherContext";
    if (to == edm::TerminationOrigin::ExternalSignal)
      context = " FromExternalSignal";
    edm::LogWarning("FastMonitoringService")
        << " STREAM " << sc.streamID().value() << " earlyTermination -: ID:" << sc.eventID()
        << " LS:" << sc.eventID().luminosityBlock() << " " << context;
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    exceptionInLS_.push_back(sc.eventID().luminosityBlock());
    has_data_exception_.store(true);
  }

preStreamEndLumi()

void evf::FastMonitoringService::preStreamEndLumi

(

edm::StreamContext const &

sc

)

Definition at line 624 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), fmt_, evf::getmEoL(), CommonMethods::lock(), edm::EventID::luminosityBlock(), microstate_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

                                                                         {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    //update processed count to be complete at this time
    //doStreamEOLSnapshot(sc.eventID().luminosityBlock(), sid);
    fmt_->jsonMonitor_->snapStreamAtomic(sc.eventID().luminosityBlock(), sc.streamID().value());
    //reset this in case stream does not get notified of next lumi (we keep processed events only)
    microstate_[sc.streamID().value()] = getmEoL();
  }

reportLockWait()

void evf::FastMonitoringService::reportLockWait	(	unsigned int	ls,
		double	waitTime,
		unsigned int	lockCount
	)

Definition at line 767 of file FastMonitoringService.cc.

References fmt_, CommonMethods::lock(), lockStatsDuringLumi_, and eostools::ls().

Referenced by FedRawDataInputSource::readSupervisor(), and DAQSource::readSupervisor().

                                                                                                     {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    lockStatsDuringLumi_[ls] = std::pair<double, unsigned int>(waitTime, lockCount);
  }

setExceptionDetected()

void evf::FastMonitoringService::setExceptionDetected

(

unsigned int

ls

)

Definition at line 449 of file FastMonitoringService.cc.

References exception_detected_, exceptionInLS_, fmt_, CommonMethods::lock(), and eostools::ls().

Referenced by FedRawDataInputSource::getNextEvent(), and DAQSource::getNextEventFromDataBlock().

                                                                  {
    std::lock_guard<std::mutex> lock(fmt_->monlock_);
    if (!ls)
      exception_detected_ = true;
    else
      exceptionInLS_.push_back(ls);
  }

setInputSource() [1/2]

void evf::FastMonitoringService::setInputSource ( FedRawDataInputSource *  inputSource )

inline

Definition at line 227 of file FastMonitoringService.h.

References inputSource_.

Referenced by DAQSource::DAQSource(), and FedRawDataInputSource::FedRawDataInputSource().

{ inputSource_ = inputSource; }

setInputSource() [2/2]

void evf::FastMonitoringService::setInputSource ( DAQSource *  inputSource )

inline

Definition at line 228 of file FastMonitoringService.h.

References daqInputSource_.

{ daqInputSource_ = inputSource; }

setInState()

void evf::FastMonitoringService::setInState ( FastMonState::InputState  inputState )

inline

Definition at line 229 of file FastMonitoringService.h.

References inputState_.

Referenced by DAQSource::DAQSource(), FedRawDataInputSource::FedRawDataInputSource(), FedRawDataInputSource::setMonState(), and DAQSource::setMonState().

{ inputState_ = inputState; }

setInStateSup()

void evf::FastMonitoringService::setInStateSup ( FastMonState::InputState  inputState )

inline

Definition at line 230 of file FastMonitoringService.h.

References inputSupervisorState_.

Referenced by DAQSource::DAQSource(), FedRawDataInputSource::FedRawDataInputSource(), FedRawDataInputSource::setMonStateSup(), and DAQSource::setMonStateSup().

{ inputSupervisorState_ = inputState; }

setTMicrostate()

void evf::FastMonitoringService::setTMicrostate

(

FastMonState::Microstate

m

)

Definition at line 772 of file FastMonitoringService.cc.

References visualization-live-secondInstance_cfg::m, specialMicroStateNames, and tmicrostate_.

Referenced by IdleSourceSentry::IdleSourceSentry(), and IdleSourceSentry::~IdleSourceSentry().

                                                                     {
    tmicrostate_[tbb::this_task_arena::current_thread_index()] = &specialMicroStateNames[m];
  }

shouldWriteFiles()

bool evf::FastMonitoringService::shouldWriteFiles ( unsigned int  lumi, unsigned int *  proc = nullptr  )

inline

Definition at line 220 of file FastMonitoringService.h.

References getAbortFlagForLumi(), getEventsProcessedForLumi(), and ValidateTausOnZEEFastSim_cff::proc.

Referenced by L1TriggerJSONMonitoring::globalEndLuminosityBlockSummary(), and HLTriggerJSONMonitoring::globalEndLuminosityBlockSummary().

                                                                           {
      unsigned int processed = getEventsProcessedForLumi(lumi);
      if (proc)
        *proc = processed;
      return !getAbortFlagForLumi(lumi);
    }

snapshotRunner()

void evf::FastMonitoringService::snapshotRunner ( )

private

Definition at line 811 of file FastMonitoringService.cc.

References doSnapshot(), f, fastMonIntervals_, fastPath_, fmt_, mps_fire::i, inputState_, inputStateNames, inputSupervisorState_, lastGlobalLumi_, CommonMethods::lock(), monInit_, mps_check::msg, nMonThreads_, AlCaHLTBitMon_ParallelJobs::p, sleepTime_, snapCounter_, and verbose_.

Referenced by preBeginJob().

                                             {
    monInit_.exchange(true, std::memory_order_acquire);
    while (!fmt_->m_stoprequest) {
      std::vector<std::vector<unsigned int>> lastEnc;
      {
        std::unique_lock<std::mutex> lock(fmt_->monlock_);

        doSnapshot(lastGlobalLumi_, false);

        lastEnc.emplace_back(fmt_->m_data.tmicrostateEncoded_);
        lastEnc.emplace_back(fmt_->m_data.microstateEncoded_);

        if (fastMonIntervals_ && (snapCounter_ % fastMonIntervals_) == 0) {
          std::vector<std::string> CSVv;
          for (unsigned int i = 0; i < nMonThreads_; i++) {
            CSVv.push_back(fmt_->jsonMonitor_->getCSVString((int)i));
          }
          // release mutex before writing out fast path file
          lock.release()->unlock();
          fmt_->jsonMonitor_->outputCSV(fastPath_, CSVv);
        }
        snapCounter_++;
      }

      if (verbose_) {
        edm::LogInfo msg("FastMonitoringService");
        auto f = [&](std::vector<unsigned int> const& p) {
          for (unsigned int i = 0; i < nMonThreads_; i++) {
            if (i == 0)
              msg << "[" << p[i] << ",";
            else if (i <= nMonThreads_ - 1)
              msg << p[i] << ",";
            else
              msg << p[i] << "]";
          }
        };

        msg << "Current states: Ms=" << fmt_->m_data.fastMacrostateJ_.value() << " ms=";
        f(lastEnc[0]);
        msg << " us=";
        f(lastEnc[1]);
        msg << " is=" << inputStateNames[inputState_] << " iss=" << inputStateNames[inputSupervisorState_];
      }

      ::sleep(sleepTime_);
    }
  }

startedLookingForFile()

void evf::FastMonitoringService::startedLookingForFile

(

)

Definition at line 731 of file FastMonitoringService.cc.

References fileLookStart_.

Referenced by FedRawDataInputSource::readSupervisor(), and DAQSource::readSupervisor().

                                                    {
    gettimeofday(&fileLookStart_, nullptr);
    /*
     std::cout << "Started looking for .raw file at: s=" << fileLookStart_.tv_sec << ": ms = "
     << fileLookStart_.tv_usec / 1000.0 << std::endl;
     */
  }

stoppedLookingForFile()

void evf::FastMonitoringService::stoppedLookingForFile

(

unsigned int

lumi

)

Definition at line 739 of file FastMonitoringService.cc.

References avgLeadTime_, fileLookStart_, fileLookStop_, fmt_, mps_fire::i, leadTimes_, CommonMethods::lock(), BXlumiParameters_cfi::lumi, and lumiFromSource_.

Referenced by FedRawDataInputSource::readSupervisor(), and DAQSource::readSupervisor().

                                                                     {
    gettimeofday(&fileLookStop_, nullptr);
    /*
     std::cout << "Stopped looking for .raw file at: s=" << fileLookStop_.tv_sec << ": ms = "
     << fileLookStop_.tv_usec / 1000.0 << std::endl;
     */
    std::lock_guard<std::mutex> lock(fmt_->monlock_);

    if (lumi > lumiFromSource_) {
      lumiFromSource_ = lumi;
      leadTimes_.clear();
    }
    unsigned long elapsedTime = (fileLookStop_.tv_sec - fileLookStart_.tv_sec) * 1000000  // sec to us
                                + (fileLookStop_.tv_usec - fileLookStart_.tv_usec);       // us
    // add this to lead times for this lumi
    leadTimes_.push_back((double)elapsedTime);

    // recompute average lead time for this lumi
    if (leadTimes_.size() == 1)
      avgLeadTime_[lumi] = leadTimes_[0];
    else {
      double totTime = 0;
      for (unsigned int i = 0; i < leadTimes_.size(); i++)
        totTime += leadTimes_[i];
      avgLeadTime_[lumi] = 0.001 * (totTime / leadTimes_.size());
    }
  }

Member Data Documentation

accuSize_

std::map<unsigned int, unsigned long> evf::FastMonitoringService::accuSize_

private

Definition at line 280 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), and preGlobalEndLumi().

avgLeadTime_

std::map<unsigned int, double> evf::FastMonitoringService::avgLeadTime_

private

Definition at line 277 of file FastMonitoringService.h.

Referenced by doSnapshot(), postGlobalEndLumi(), and stoppedLookingForFile().

ct_

std::unique_ptr<ConcurrencyTracker> evf::FastMonitoringService::ct_

private

Definition at line 247 of file FastMonitoringService.h.

Referenced by doSnapshot(), preallocate(), and preBeginJob().

daqInputSource_

DAQSource* evf::FastMonitoringService::daqInputSource_ = nullptr

private

Definition at line 251 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

exception_detected_

bool evf::FastMonitoringService::exception_detected_ = false

private

Definition at line 303 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), preSourceEarlyTermination(), and setExceptionDetected().

exceptionInLS_

std::vector<unsigned int> evf::FastMonitoringService::exceptionInLS_

private

Definition at line 306 of file FastMonitoringService.h.

Referenced by isExceptionOnData(), preGlobalEarlyTermination(), preGlobalEndLumi(), preStreamEarlyTermination(), and setExceptionDetected().

fastMicrostateDefPath_

std::string evf::FastMonitoringService::fastMicrostateDefPath_

private

Definition at line 263 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

fastMonIntervals_

unsigned int evf::FastMonitoringService::fastMonIntervals_

private

Definition at line 261 of file FastMonitoringService.h.

Referenced by snapshotRunner().

fastName_

std::string evf::FastMonitoringService::fastName_

private

Definition at line 264 of file FastMonitoringService.h.

Referenced by preBeginJob().

fastPath_

std::string evf::FastMonitoringService::fastPath_

private

Definition at line 264 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

fileLookStart_

timeval evf::FastMonitoringService::fileLookStart_

private

Definition at line 269 of file FastMonitoringService.h.

Referenced by startedLookingForFile(), and stoppedLookingForFile().

fileLookStop_

timeval evf::FastMonitoringService::fileLookStop_

private

Definition at line 269 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

filesProcessedDuringLumi_

std::map<unsigned int, unsigned int> evf::FastMonitoringService::filesProcessedDuringLumi_

private

Definition at line 278 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), doSnapshot(), and postGlobalEndLumi().

fmt_

std::unique_ptr<FastMonitoringThread> evf::FastMonitoringService::fmt_

private

Definition at line 246 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), doSnapshot(), getAbortFlagForLumi(), getEventsProcessedForLumi(), isExceptionOnData(), jobFailure(), makeModuleLegendaJson(), postBeginJob(), postEndJob(), postEvent(), postGlobalBeginRun(), postGlobalEndLumi(), preBeginJob(), preGlobalBeginLumi(), preGlobalEarlyTermination(), preGlobalEndLumi(), preModuleBeginJob(), preSourceEarlyTermination(), preStreamBeginLumi(), preStreamEarlyTermination(), preStreamEndLumi(), reportLockWait(), setExceptionDetected(), snapshotRunner(), and stoppedLookingForFile().

has_data_exception_

std::atomic<bool> evf::FastMonitoringService::has_data_exception_ = false

private

Definition at line 305 of file FastMonitoringService.h.

Referenced by exceptionDetected(), isExceptionOnData(), preGlobalEarlyTermination(), preSourceEarlyTermination(), and preStreamEarlyTermination().

has_source_exception_

std::atomic<bool> evf::FastMonitoringService::has_source_exception_ = false

private

Definition at line 304 of file FastMonitoringService.h.

Referenced by exceptionDetected(), isExceptionOnData(), and preSourceEarlyTermination().

inputLegendFileJson_

std::string evf::FastMonitoringService::inputLegendFileJson_

private

Definition at line 299 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

inputSource_

FedRawDataInputSource* evf::FastMonitoringService::inputSource_ = nullptr

private

Definition at line 250 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

inputState_

std::atomic<FastMonState::InputState> evf::FastMonitoringService::inputState_ {FastMonState::InputState::inInit}

private

Definition at line 252 of file FastMonitoringService.h.

Referenced by doSnapshot(), setInState(), and snapshotRunner().

inputStateNames

const std::string evf::FastMonitoringService::inputStateNames

static

Definition at line 172 of file FastMonitoringService.h.

Referenced by makeInputLegendaJson(), and snapshotRunner().

inputSupervisorState_

std::atomic<FastMonState::InputState> evf::FastMonitoringService::inputSupervisorState_ {FastMonState::InputState::inInit}

private

Definition at line 253 of file FastMonitoringService.h.

Referenced by doSnapshot(), setInStateSup(), and snapshotRunner().

isInitTransition_

std::atomic<bool> evf::FastMonitoringService::isInitTransition_

private

Definition at line 272 of file FastMonitoringService.h.

Referenced by doSnapshot(), postGlobalBeginRun(), and preBeginJob().

lastGlobalLumi_

unsigned int evf::FastMonitoringService::lastGlobalLumi_

private

Definition at line 271 of file FastMonitoringService.h.

Referenced by preBeginJob(), preGlobalBeginLumi(), and snapshotRunner().

leadTimes_

std::vector<double> evf::FastMonitoringService::leadTimes_

private

Definition at line 281 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

lockStatsDuringLumi_

std::map<unsigned int, std::pair<double, unsigned int> > evf::FastMonitoringService::lockStatsDuringLumi_

private

Definition at line 282 of file FastMonitoringService.h.

Referenced by doSnapshot(), postGlobalEndLumi(), and reportLockWait().

lumiFromSource_

unsigned int evf::FastMonitoringService::lumiFromSource_

private

Definition at line 273 of file FastMonitoringService.h.

Referenced by preBeginJob(), and stoppedLookingForFile().

lumiStartTime_

std::map<unsigned int, timeval> evf::FastMonitoringService::lumiStartTime_

private

Definition at line 268 of file FastMonitoringService.h.

Referenced by preGlobalBeginLumi(), and preGlobalEndLumi().

macroStateNames

const std::string evf::FastMonitoringService::macroStateNames

static

Initial value:

= {"Init",
                                                                                    "JobReady",
                                                                                    "RunGiven",
                                                                                    "Running",
                                                                                    "Stopping",
                                                                                    "Done",
                                                                                    "JobEnded",
                                                                                    "Error",
                                                                                    "ErrorEnded",
                                                                                    "End",
                                                                                    "Invalid"}

Definition at line 171 of file FastMonitoringService.h.

microstate_

std::vector<ContainableAtomic<const void*> > evf::FastMonitoringService::microstate_

private

Definition at line 309 of file FastMonitoringService.h.

Referenced by doSnapshot(), postEvent(), postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), postStreamBeginLumi(), postStreamEndLumi(), preBeginJob(), preEvent(), preModuleEvent(), preModuleEventAcquire(), preSourceEvent(), preStreamBeginLumi(), and preStreamEndLumi().

microstateAcqFlag_

std::vector<ContainableAtomic<unsigned char> > evf::FastMonitoringService::microstateAcqFlag_

private

Definition at line 310 of file FastMonitoringService.h.

Referenced by doSnapshot(), postModuleEventAcquire(), preBeginJob(), and preModuleEventAcquire().

microstateDefPath_

std::string evf::FastMonitoringService::microstateDefPath_

private

Definition at line 263 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

moduleLegendFile_

std::string evf::FastMonitoringService::moduleLegendFile_

private

Definition at line 297 of file FastMonitoringService.h.

Referenced by preBeginJob().

moduleLegendFileJson_

std::string evf::FastMonitoringService::moduleLegendFileJson_

private

Definition at line 298 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

monInit_

std::atomic<bool> evf::FastMonitoringService::monInit_

private

Definition at line 302 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

nMonThreads_

unsigned int evf::FastMonitoringService::nMonThreads_ = 0

private

Definition at line 256 of file FastMonitoringService.h.

Referenced by doSnapshot(), preallocate(), preBeginJob(), and snapshotRunner().

nOutputModules_

unsigned int evf::FastMonitoringService::nOutputModules_ = 0

private

Definition at line 300 of file FastMonitoringService.h.

Referenced by makeModuleLegendaJson(), and preModuleBeginJob().

nStreams_

unsigned int evf::FastMonitoringService::nStreams_ = 0

private

Definition at line 255 of file FastMonitoringService.h.

Referenced by doSnapshot(), preallocate(), and preBeginJob().

nThreads_

unsigned int evf::FastMonitoringService::nThreads_ = 0

private

Definition at line 257 of file FastMonitoringService.h.

Referenced by doSnapshot(), postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), preallocate(), preBeginJob(), preGlobalBeginLumi(), preModuleEvent(), preModuleEventAcquire(), and preSourceEvent().

processedEventsPerLumi_

std::map<unsigned int, std::pair<unsigned int, bool> > evf::FastMonitoringService::processedEventsPerLumi_

private

Definition at line 285 of file FastMonitoringService.h.

Referenced by getAbortFlagForLumi(), getEventsProcessedForLumi(), postGlobalEndLumi(), and preGlobalEndLumi().

runDirectory_

std::filesystem::path evf::FastMonitoringService::runDirectory_

private

Definition at line 291 of file FastMonitoringService.h.

Referenced by getRunDirName(), and preBeginJob().

sleepTime_

int evf::FastMonitoringService::sleepTime_

private

Definition at line 260 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

snapCounter_

unsigned int evf::FastMonitoringService::snapCounter_ = 0

private

Definition at line 262 of file FastMonitoringService.h.

Referenced by snapshotRunner().

specialMicroStateNames

const edm::ModuleDescription evf::FastMonitoringService::specialMicroStateNames

static

Initial value:

= {
      edm::ModuleDescription("Dummy", "Invalid"),
      edm::ModuleDescription("Dummy", "Idle"),
      edm::ModuleDescription("Dummy", "FwkOvhSrc"),
      edm::ModuleDescription("Dummy", "FwkOvhMod"),  
      edm::ModuleDescription("Dummy", "FwkEoL"),
      edm::ModuleDescription("Dummy", "Input"),
      edm::ModuleDescription("Dummy", "DQM"),
      edm::ModuleDescription("Dummy", "BoL"),
      edm::ModuleDescription("Dummy", "EoL"),
      edm::ModuleDescription("Dummy", "GlobalEoL"),
      edm::ModuleDescription("Dummy", "Fwk"),
      edm::ModuleDescription("Dummy", "IdleSource"),
      edm::ModuleDescription("Dummy", "Event"),
      edm::ModuleDescription("Dummy", "Ignore")}

Definition at line 170 of file FastMonitoringService.h.

Referenced by preBeginJob(), and setTMicrostate().

streamCounterUpdating_

std::vector<std::atomic<bool>*> evf::FastMonitoringService::streamCounterUpdating_

private

Definition at line 289 of file FastMonitoringService.h.

Referenced by preBeginJob().

tbbConcurrencyTracker_

bool evf::FastMonitoringService::tbbConcurrencyTracker_

private

Definition at line 259 of file FastMonitoringService.h.

Referenced by preBeginJob().

tbbMonitoringMode_

bool evf::FastMonitoringService::tbbMonitoringMode_

private

Definition at line 258 of file FastMonitoringService.h.

Referenced by postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), preGlobalBeginLumi(), preModuleEvent(), preModuleEventAcquire(), and preSourceEvent().

threadIDAvailable_

bool evf::FastMonitoringService::threadIDAvailable_ = false

private

Definition at line 293 of file FastMonitoringService.h.

tmicrostate_

std::vector<ContainableAtomic<const void*> > evf::FastMonitoringService::tmicrostate_

private

Definition at line 312 of file FastMonitoringService.h.

Referenced by doSnapshot(), postModuleEvent(), postModuleEventAcquire(), postSourceEvent(), preBeginJob(), preGlobalBeginLumi(), preModuleEvent(), preModuleEventAcquire(), preSourceEvent(), and setTMicrostate().

tmicrostateAcqFlag_

std::vector<ContainableAtomic<unsigned char> > evf::FastMonitoringService::tmicrostateAcqFlag_

private

Definition at line 313 of file FastMonitoringService.h.

Referenced by doSnapshot(), postModuleEventAcquire(), preBeginJob(), and preModuleEventAcquire().

totalEventsProcessed_

std::atomic<unsigned long> evf::FastMonitoringService::totalEventsProcessed_

private

Definition at line 295 of file FastMonitoringService.h.

Referenced by postEvent().

verbose_

bool evf::FastMonitoringService::verbose_ = false

private

Definition at line 315 of file FastMonitoringService.h.

Referenced by snapshotRunner().

workingDirectory_

std::filesystem::path evf::FastMonitoringService::workingDirectory_

private

Definition at line 291 of file FastMonitoringService.h.

Referenced by preBeginJob().