evf::FastMonitoringService Class Reference

#include <FastMonitoringService.h>

Inheritance diagram for evf::FastMonitoringService:

Classes
struct	Encoding

Public Member Functions
void	accumulateFileSize (unsigned int lumi, unsigned long fileSize)
	FastMonitoringService (const edm::ParameterSet &, edm::ActivityRegistry &)
bool	getAbortFlagForLumi (unsigned int lumi)
unsigned int	getEventsProcessedForLumi (unsigned int lumi, bool *abortFlag=0)
std::string	getRunDirName () const
void	jobFailure ()
std::string	makeInputLegendaJson ()
std::string	makeModuleLegendaJson ()
std::string	makePathLegendaJson ()
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	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	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	setInState (FastMonitoringThread::InputState inputState)
void	setInStateSup (FastMonitoringThread::InputState inputState)
void	setMicroState (MicroStateService::Microstate)
void	setMicroState (edm::StreamID, MicroStateService::Microstate)
bool	shouldWriteFiles (unsigned int lumi, unsigned int *proc=0)
void	startedLookingForFile ()
void	stoppedLookingForFile (unsigned int lumi)
	~FastMonitoringService ()
Public Member Functions inherited from evf::MicroStateService
virtual std::string	getMicroState1 ()
virtual std::string const &	getMicroState2 ()
	MicroStateService (const edm::ParameterSet &, edm::ActivityRegistry &)
virtual	~MicroStateService ()

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

Static Public Attributes
static const std::string	inputStateNames [FastMonitoringThread::inCOUNT]
static const std::string	macroStateNames [FastMonitoringThread::MCOUNT]
static const std::string	nopath_ = "NoPath"
Static Public Attributes inherited from evf::MicroStateService
static const edm::ModuleDescription	reservedMicroStateNames [mCOUNT]

Private Member Functions
void	doSnapshot (const unsigned int ls, const bool isGlobalEOL)
void	doStreamEOLSnapshot (const unsigned int ls, const unsigned int streamID)
void	dowork ()

Private Attributes
std::map< unsigned int, unsigned long >	accuSize_
std::map< unsigned int, double >	avgLeadTime_
std::vector< std::atomic< bool > * >	collectedPathList_
bool	emptyLumisectionMode_ = false
Encoding	encModule_
std::vector< Encoding >	encPath_
std::vector< unsigned int >	eventCountForPathInit_
bool	exception_detected_ = false
std::vector< unsigned int >	exceptionInLS_
std::string	fastMicrostateDefPath_
unsigned int	fastMonIntervals_
std::string	fastName_
std::string	fastPath_
std::vector< std::string >	fastPathList_
timeval	fileLookStart_
timeval	fileLookStop_
bool	filePerFwkStream_
std::map< unsigned int, unsigned int >	filesProcessedDuringLumi_
std::vector< unsigned long >	firstEventId_
FastMonitoringThread	fmt_
std::string	inputLegendFileJson_
FedRawDataInputSource *	inputSource_ = 0
FastMonitoringThread::InputState	inputState_
FastMonitoringThread::InputState	inputSupervisorState_
bool	isGlobalLumiTransition_
bool	isInitTransition_
unsigned int	lastGlobalLumi_
std::queue< unsigned int >	lastGlobalLumisClosed_
std::vector< double >	leadTimes_
std::map< unsigned int, std::pair< double, unsigned int > >	lockStatsDuringLumi_
unsigned int	lumiFromSource_
std::map< unsigned int, timeval >	lumiStartTime_
FastMonitoringThread::Macrostate	macrostate_
std::vector< const void * >	microstate_
std::string	microstateDefPath_
std::vector< const void * >	ministate_
std::string	moduleLegendFile_
std::string	moduleLegendFileJson_
std::atomic< bool >	monInit_
unsigned int	nOutputModules_ =0
unsigned int	nStreams_
unsigned int	nThreads_
std::string	pathLegendFile_
std::string	pathLegendFileJson_
bool	pathLegendWritten_ = false
std::vector< bool >	pathNamesReady_
std::map< unsigned int, std::pair< unsigned int, bool > >	processedEventsPerLumi_
boost::filesystem::path	runDirectory_
int	sleepTime_
std::string	slowName_
unsigned int	snapCounter_ = 0
std::vector< std::atomic< bool > * >	streamCounterUpdating_
bool	threadIDAvailable_ = false
std::vector< const void * >	threadMicrostate_
std::atomic< unsigned long >	totalEventsProcessed_
boost::filesystem::path	workingDirectory_

Additional Inherited Members
Public Types inherited from evf::MicroStateService
enum	Microstate {   mInvalid = 0, mIdle, mFwkOvhSrc, mFwkOvhMod,   mFwkEoL, mInput, mDqm, mBoL,   mEoL, mGlobEoL, mCOUNT }
Static Protected Attributes inherited from evf::MicroStateService
static const std::string	default_return_ ="NotImplemented"

Detailed Description

Definition at line 57 of file FastMonitoringService.h.

Constructor & Destructor Documentation

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

Definition at line 60 of file FastMonitoringService.cc.

References Exception, fastMicrostateDefPath_, jobFailure(), microstateDefPath_, postBeginJob(), postEndJob(), postEvent(), postGlobalEndLumi(), postModuleEvent(), postSourceEvent(), postStreamBeginLumi(), postStreamEndLumi(), preallocate(), preBeginJob(), preEvent(), preGlobalBeginLumi(), preGlobalEarlyTermination(), preGlobalEndLumi(), preModuleBeginJob(), preModuleEvent(), prePathEvent(), preSourceEarlyTermination(), preSourceEvent(), preStreamBeginLumi(), preStreamEarlyTermination(), preStreamEndLumi(), AlCaHLTBitMon_QueryRunRegistry::string, edm::ActivityRegistry::watchJobFailure(), edm::ActivityRegistry::watchPostBeginJob(), edm::ActivityRegistry::watchPostEndJob(), edm::ActivityRegistry::watchPostEvent(), edm::ActivityRegistry::watchPostGlobalEndLumi(), edm::ActivityRegistry::watchPostModuleEvent(), 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::watchPrePathEvent(), edm::ActivityRegistry::watchPreSourceEarlyTermination(), edm::ActivityRegistry::watchPreSourceEvent(), edm::ActivityRegistry::watchPreStreamBeginLumi(), edm::ActivityRegistry::watchPreStreamEarlyTermination(), and edm::ActivityRegistry::watchPreStreamEndLumi().

                                                 : 
    MicroStateService(iPS,reg)
    ,encModule_(nReservedModules)
    ,nStreams_(0)//until initialized
    ,sleepTime_(iPS.getUntrackedParameter<int>("sleepTime", 1))
    ,fastMonIntervals_(iPS.getUntrackedParameter<unsigned int>("fastMonIntervals", 2))
    ,fastName_("fastmoni")
    ,slowName_("slowmoni")
    ,filePerFwkStream_(iPS.getUntrackedParameter<bool>("filePerFwkStream", false))
    ,totalEventsProcessed_(0)
  {
    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.watchPrePathEvent(this,&FastMonitoringService::prePathEvent);

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

    reg.watchPreSourceEvent(this,&FastMonitoringService::preSourceEvent);//source (with streamID of requestor)
    reg.watchPostSourceEvent(this,&FastMonitoringService::postSourceEvent);
  
    reg.watchPreModuleEvent(this,&FastMonitoringService::preModuleEvent);//should be 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(getenv("CMSSW_BASE")) + "/" + microstateBaseSuffix;
    if (stat(microstatePath.c_str(), &statbuf)) {
      microstatePath = std::string(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;
  }

evf::FastMonitoringService::~FastMonitoringService

(

)

Definition at line 120 of file FastMonitoringService.cc.

  {
  }

Member Function Documentation

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

Definition at line 655 of file FastMonitoringService.cc.

References accuSize_, filesProcessedDuringLumi_, fmt_, CommonMethods::lock(), fjr2json::lumi, and evf::FastMonitoringThread::monlock_.

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

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

private

Definition at line 739 of file FastMonitoringService.cc.

References avgLeadTime_, encModule_, evf::FastMonitoringService::Encoding::encode(), encPath_, evf::FastMonitoringThread::MonitorData::fastAvgLeadTimeJ_, evf::FastMonitoringThread::MonitorData::fastFilesProcessedJ_, evf::FastMonitoringThread::MonitorData::fastLockCountJ_, evf::FastMonitoringThread::MonitorData::fastLockWaitJ_, evf::FastMonitoringThread::MonitorData::fastMacrostateJ_, filesProcessedDuringLumi_, fmt_, i, reco::if(), evf::FastMonitoringThread::inNewLumi, evf::FastMonitoringThread::inNewLumiBusyEndingLS, evf::FastMonitoringThread::inNewLumiIdleEndingLS, evf::FastMonitoringThread::inNoRequest, evf::FastMonitoringThread::inNoRequestWithEoLThreads, evf::FastMonitoringThread::inNoRequestWithGlobalEoL, evf::FastMonitoringThread::inNoRequestWithIdleThreads, evf::FastMonitoringThread::MonitorData::inputState_, inputState_, inputSupervisorState_, evf::FastMonitoringThread::inRunEnd, evf::FastMonitoringThread::inSupBusy, evf::FastMonitoringThread::inSupFileLimit, evf::FastMonitoringThread::inSupLockPolling, evf::FastMonitoringThread::inSupLockPollingCopying, 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, evf::FastMonitoringThread::inWaitChunk, evf::FastMonitoringThread::inWaitChunk_busy, evf::FastMonitoringThread::inWaitChunk_fileLimit, evf::FastMonitoringThread::inWaitChunk_lockPolling, evf::FastMonitoringThread::inWaitChunk_lockPollingCopying, evf::FastMonitoringThread::inWaitChunk_newFile, evf::FastMonitoringThread::inWaitChunk_newFileWaitChunk, evf::FastMonitoringThread::inWaitChunk_newFileWaitChunkCopying, evf::FastMonitoringThread::inWaitChunk_newFileWaitThread, evf::FastMonitoringThread::inWaitChunk_newFileWaitThreadCopying, evf::FastMonitoringThread::inWaitChunk_noFile, evf::FastMonitoringThread::inWaitChunk_runEnd, evf::FastMonitoringThread::inWaitChunk_waitFreeChunk, evf::FastMonitoringThread::inWaitChunk_waitFreeChunkCopying, evf::FastMonitoringThread::inWaitChunk_waitFreeThread, evf::FastMonitoringThread::inWaitChunk_waitFreeThreadCopying, evf::FastMonitoringThread::inWaitInput, evf::FastMonitoringThread::inWaitInput_busy, evf::FastMonitoringThread::inWaitInput_fileLimit, evf::FastMonitoringThread::inWaitInput_lockPolling, evf::FastMonitoringThread::inWaitInput_lockPollingCopying, evf::FastMonitoringThread::inWaitInput_newFile, evf::FastMonitoringThread::inWaitInput_newFileWaitChunk, evf::FastMonitoringThread::inWaitInput_newFileWaitChunkCopying, evf::FastMonitoringThread::inWaitInput_newFileWaitThread, evf::FastMonitoringThread::inWaitInput_newFileWaitThreadCopying, evf::FastMonitoringThread::inWaitInput_noFile, evf::FastMonitoringThread::inWaitInput_runEnd, evf::FastMonitoringThread::inWaitInput_waitFreeChunk, evf::FastMonitoringThread::inWaitInput_waitFreeChunkCopying, evf::FastMonitoringThread::inWaitInput_waitFreeThread, evf::FastMonitoringThread::inWaitInput_waitFreeThreadCopying, isGlobalLumiTransition_, isInitTransition_, evf::FastMonitoringThread::jsonMonitor_, lockStatsDuringLumi_, evf::FastMonitoringThread::m_data, macrostate_, evf::MicroStateService::mEoL, evf::MicroStateService::mFwkEoL, evf::MicroStateService::mGlobEoL, microstate_, evf::FastMonitoringThread::MonitorData::microstateEncoded_, evf::MicroStateService::mIdle, ministate_, evf::FastMonitoringThread::MonitorData::ministateEncoded_, nStreams_, and evf::MicroStateService::reservedMicroStateNames.

Referenced by dowork(), and preGlobalEndLumi().

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

    //update these unless in the midst of a global transition
    if (!isGlobalLumiTransition_ && !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.;
      }
    }
    else {
      if (isGlobalLumiTransition_)
        for (unsigned int i=0;i<nStreams_;i++) {
          if (microstate_[i]==&reservedMicroStateNames[mFwkEoL]) {
            microstate_[i]=&reservedMicroStateNames[mGlobEoL];
          }
        }
    }

    for (unsigned int i=0;i<nStreams_;i++) {
      fmt_.m_data.ministateEncoded_[i] = encPath_[i].encode(ministate_[i]);
      fmt_.m_data.microstateEncoded_[i] = encModule_.encode(microstate_[i]);
    }

    bool inputStatePerThread=false;

    if (inputState_==FastMonitoringThread::inWaitInput) {
      switch (inputSupervisorState_) {
        case FastMonitoringThread::inSupFileLimit:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_fileLimit;
          break;
        case FastMonitoringThread::inSupWaitFreeChunk:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_waitFreeChunk;
          break;
        case FastMonitoringThread::inSupWaitFreeChunkCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_waitFreeChunkCopying;
          break;
        case FastMonitoringThread::inSupWaitFreeThread:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_waitFreeThread;
          break;
        case FastMonitoringThread::inSupWaitFreeThreadCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_waitFreeThreadCopying;
          break;
        case FastMonitoringThread::inSupBusy:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_busy;
          break;
        case FastMonitoringThread::inSupLockPolling:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_lockPolling;
          break;
        case FastMonitoringThread::inSupLockPollingCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_lockPollingCopying;
          break;
        case FastMonitoringThread::inRunEnd:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_runEnd;
          break;
        case FastMonitoringThread::inSupNoFile:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_noFile;
          break;
        case FastMonitoringThread::inSupNewFile:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_newFile;
          break;
        case FastMonitoringThread::inSupNewFileWaitThreadCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_newFileWaitThreadCopying;
          break;
        case FastMonitoringThread::inSupNewFileWaitThread:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_newFileWaitThread;
          break;
        case FastMonitoringThread::inSupNewFileWaitChunkCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_newFileWaitChunkCopying;
          break;
        case FastMonitoringThread::inSupNewFileWaitChunk:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput_newFileWaitChunk;
          break;
        default: 
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitInput;
      }
    }
    else if (inputState_==FastMonitoringThread::inWaitChunk) {

      switch (inputSupervisorState_) {
        case FastMonitoringThread::inSupFileLimit:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_fileLimit;
          break;
        case FastMonitoringThread::inSupWaitFreeChunk:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_waitFreeChunk;
          break;
        case FastMonitoringThread::inSupWaitFreeChunkCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_waitFreeChunkCopying;
          break;
        case FastMonitoringThread::inSupWaitFreeThread:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_waitFreeThread;
          break;
        case FastMonitoringThread::inSupWaitFreeThreadCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_waitFreeThreadCopying;
          break;
        case FastMonitoringThread::inSupBusy:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_busy;
          break;
        case FastMonitoringThread::inSupLockPolling:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_lockPolling;
          break;
        case FastMonitoringThread::inSupLockPollingCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_lockPollingCopying;
          break;
        case FastMonitoringThread::inRunEnd:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_runEnd;
          break;
        case FastMonitoringThread::inSupNoFile:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_noFile;
          break;
        case FastMonitoringThread::inSupNewFile:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_newFile;
          break;
        case FastMonitoringThread::inSupNewFileWaitThreadCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_newFileWaitThreadCopying;
          break;
        case FastMonitoringThread::inSupNewFileWaitThread:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_newFileWaitThread;
          break;
        case FastMonitoringThread::inSupNewFileWaitChunkCopying:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_newFileWaitChunkCopying;
          break;
        case FastMonitoringThread::inSupNewFileWaitChunk:
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk_newFileWaitChunk;
          break;
        default: 
          fmt_.m_data.inputState_[0]=FastMonitoringThread::inWaitChunk;
      }
    }
    else if (inputState_==FastMonitoringThread::inNoRequest) {
      if (isGlobalLumiTransition_ && !isInitTransition_)
        fmt_.m_data.inputState_[0]=FastMonitoringThread::inNoRequestWithGlobalEoL;
      else {
        inputStatePerThread=true;
        for (unsigned int i=0;i<nStreams_;i++) {
          if (microstate_[i]==&reservedMicroStateNames[mIdle])
            fmt_.m_data.inputState_[i]=FastMonitoringThread::inNoRequestWithIdleThreads;
          else if (microstate_[i]==&reservedMicroStateNames[mEoL] || 
            microstate_[i]==&reservedMicroStateNames[mFwkEoL] || 
            microstate_[i]==&reservedMicroStateNames[mGlobEoL])
            fmt_.m_data.inputState_[i]=FastMonitoringThread::inNoRequestWithEoLThreads;
          else
            fmt_.m_data.inputState_[i]=FastMonitoringThread::inNoRequest;
        }
      }
    }
    else if (inputState_ == FastMonitoringThread::inNewLumi) {
      inputStatePerThread=true;
      for (unsigned int i=0;i<nStreams_;i++) {
        if (microstate_[i]==&reservedMicroStateNames[mEoL] || 
          microstate_[i]==&reservedMicroStateNames[mFwkEoL] || 
          microstate_[i]==&reservedMicroStateNames[mGlobEoL])
          fmt_.m_data.inputState_[i]=FastMonitoringThread::inNewLumi;
        else if (microstate_[i]==&reservedMicroStateNames[mIdle])
          fmt_.m_data.inputState_[i]=FastMonitoringThread::inNewLumiIdleEndingLS;
        else
          fmt_.m_data.inputState_[i]=FastMonitoringThread::inNewLumiBusyEndingLS;
      }
    }
    else
      fmt_.m_data.inputState_[0]=inputState_;

    //this is same for all streams
    if (!inputStatePerThread)
      for (unsigned int i=1;i<nStreams_;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);
  }

void evf::FastMonitoringService::doStreamEOLSnapshot ( const unsigned int  ls, const unsigned int  streamID  )

inlineprivate

Definition at line 181 of file FastMonitoringService.h.

References fmt_, and evf::FastMonitoringThread::jsonMonitor_.

Referenced by preStreamEndLumi().

                                                                                   {
    //pick up only event count here
    fmt_.jsonMonitor_->snapStreamAtomic(ls,streamID);
      }

void evf::FastMonitoringService::dowork ( )

inlineprivate

Definition at line 186 of file FastMonitoringService.h.

References doSnapshot(), encModule_, evf::FastMonitoringService::Encoding::encode(), encPath_, evf::FastMonitoringThread::MonitorData::fastMacrostateJ_, fastMonIntervals_, fastPath_, fastPathList_, filePerFwkStream_, fmt_, i, inputState_, inputStateNames, inputSupervisorState_, evf::FastMonitoringThread::jsonMonitor_, lastGlobalLumi_, CommonMethods::lock(), evf::FastMonitoringThread::m_data, evf::FastMonitoringThread::m_stoprequest, microstate_, ministate_, monInit_, evf::FastMonitoringThread::monlock_, nStreams_, findQualityFiles::size, sleepTime_, snapCounter_, AlCaHLTBitMon_QueryRunRegistry::string, and jsoncollector::IntJ::value().

Referenced by preBeginJob().

                    { // the function to be called in the thread. Thread completes when function returns.
        monInit_.exchange(true,std::memory_order_acquire);
    while (!fmt_.m_stoprequest) {
      edm::LogInfo("FastMonitoringService") << "Current states: Ms=" << fmt_.m_data.fastMacrostateJ_.value()
                << " ms=" << encPath_[0].encode(ministate_[0])
                << " us=" << encModule_.encode(microstate_[0])
                << " is=" << inputStateNames[inputState_]
                << " iss="<< inputStateNames[inputSupervisorState_]
                    << std::endl;

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

            doSnapshot(lastGlobalLumi_,false);

            if (fastMonIntervals_ && (snapCounter_%fastMonIntervals_)==0) {
              if (filePerFwkStream_) {
                std::vector<std::string> CSVv;
                for (unsigned int i=0;i<nStreams_;i++) {
                  CSVv.push_back(fmt_.jsonMonitor_->getCSVString((int)i));
                }
                fmt_.monlock_.unlock();
                for (unsigned int i=0;i<nStreams_;i++) {
                  if (CSVv[i].size())
                    fmt_.jsonMonitor_->outputCSV(fastPathList_[i],CSVv[i]);
                }
              }
              else {
                std::string CSV = fmt_.jsonMonitor_->getCSVString();
                //release mutex before writing out fast path file
                fmt_.monlock_.unlock();
                if (CSV.size())
                  fmt_.jsonMonitor_->outputCSV(fastPath_,CSV);
              }
            }

            snapCounter_++;
            
          }
      ::sleep(sleepTime_);
    }
      }

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

static

Definition at line 124 of file FastMonitoringService.cc.

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

  {
    edm::ParameterSetDescription desc;
    desc.setComment("Service for File-based DAQ monitoring and event accounting");
    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", false)->setComment("Switches on monitoring output per framework stream");
    desc.setAllowAnything();
    descriptions.add("FastMonitoringService", desc);
  }

bool evf::FastMonitoringService::getAbortFlagForLumi

(

unsigned int

lumi

)

Definition at line 725 of file FastMonitoringService.cc.

References Exception, fmt_, CommonMethods::lock(), fjr2json::lumi, evf::FastMonitoringThread::monlock_, 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 0;
    }
  }

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

Definition at line 709 of file FastMonitoringService.cc.

References Exception, fmt_, CommonMethods::lock(), fjr2json::lumi, evf::FastMonitoringThread::monlock_, proc, and processedEventsPerLumi_.

Referenced by dqm::DQMFileSaverPB::fillJson(), DQMFileSaver::fillJson(), DQMFileSaver::saveForFilterUnit(), 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;
    }
  }

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

inline

Definition at line 172 of file FastMonitoringService.h.

References runDirectory_.

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

void evf::FastMonitoringService::jobFailure

(

)

Definition at line 347 of file FastMonitoringService.cc.

References macrostate_, and evf::FastMonitoringThread::sError.

Referenced by FastMonitoringService().

  {
    macrostate_ = FastMonitoringThread::sError;
  }

std::string evf::FastMonitoringService::makeInputLegendaJson

(

)

Definition at line 164 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, i, evf::FastMonitoringThread::inCOUNT, inputStateNames, and Json::StyledWriter::write().

Referenced by postBeginJob().

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

std::string evf::FastMonitoringService::makeModuleLegendaJson

(

)

Definition at line 148 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, evf::FastMonitoringService::Encoding::current_, evf::FastMonitoringService::Encoding::decode(), encModule_, i, nOutputModules_, nReservedModules, nSpecialModules, and Json::StyledWriter::write().

Referenced by postBeginJob().

                                                        {
    Json::Value legendaVector(Json::arrayValue);
    for(int i = 0; i < encModule_.current_; i++)
       legendaVector.append(Json::Value((static_cast<const edm::ModuleDescription *>(encModule_.decode(i)))->moduleLabel()));
    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);
  }

std::string evf::FastMonitoringService::makePathLegendaJson

(

)

Definition at line 136 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, encPath_, i, nReservedPaths, and Json::StyledWriter::write().

Referenced by prePathEvent().

                                                       {
    Json::Value legendaVector(Json::arrayValue);
    for(int i = 0; i < encPath_[0].current_; i++)
      legendaVector.append(Json::Value(*(static_cast<const std::string *>(encPath_[0].decode(i)))));
    Json::Value valReserved(nReservedPaths);
    Json::Value pathLegend;
    pathLegend["names"]=legendaVector;
    pathLegend["reserved"]=valReserved;
    Json::StyledWriter writer;
    return writer.write(pathLegend);
  }

void evf::FastMonitoringService::postBeginJob

(

)

Definition at line 369 of file FastMonitoringService.cc.

References encModule_, fmt_, inputLegendFileJson_, CommonMethods::lock(), evf::FastMonitoringThread::m_data, macrostate_, makeInputLegendaJson(), makeModuleLegendaJson(), evf::FastMonitoringThread::MonitorData::microstateBins_, moduleLegendFileJson_, evf::FastMonitoringThread::monlock_, evf::FastMonitoringThread::sJobReady, AlCaHLTBitMon_QueryRunRegistry::string, and evf::FastMonitoringService::Encoding::vecsize().

Referenced by FastMonitoringService().

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

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

    macrostate_ = FastMonitoringThread::sJobReady;

    //update number of entries in module histogram
    std::lock_guard<std::mutex> lock(fmt_.monlock_);
    fmt_.m_data.microstateBins_ = encModule_.vecsize();
  }

void evf::FastMonitoringService::postEndJob

(

)

Definition at line 384 of file FastMonitoringService.cc.

References fmt_, macrostate_, evf::FastMonitoringThread::sJobEnded, and evf::FastMonitoringThread::stop().

Referenced by FastMonitoringService().

  {
    macrostate_ = FastMonitoringThread::sJobEnded;
    fmt_.stop();
  }

void evf::FastMonitoringService::postEvent

(

edm::StreamContext const &

sc

)

Definition at line 590 of file FastMonitoringService.cc.

References eventCountForPathInit_, evf::FastMonitoringThread::MonitorData::fastPathProcessedJ_, fmt_, evf::FastMonitoringThread::m_data, microstate_, evf::MicroStateService::mIdle, ministate_, nopath_, evf::FastMonitoringThread::MonitorData::processed_, evf::MicroStateService::reservedMicroStateNames, streamCounterUpdating_, edm::StreamContext::streamID(), and totalEventsProcessed_.

Referenced by FastMonitoringService().

  {
    microstate_[sc.streamID()] = &reservedMicroStateNames[mIdle];

    ministate_[sc.streamID()] = &nopath_;

    #if ATOMIC_LEVEL>=2
    //use atomic flag to make sure end of lumi sees this
    streamCounterUpdating_[sc.streamID()]->store(true,std::memory_order_release);
    fmt_.m_data.processed_[sc.streamID()]->fetch_add(1,std::memory_order_release);
    streamCounterUpdating_[sc.streamID()]->store(false,std::memory_order_release);

    #elif ATOMIC_LEVEL==1
    //writes are atomic, we assume writes propagate to memory before stream EOL snap
    fmt_.m_data.processed_[sc.streamID()]->fetch_add(1,std::memory_order_relaxed);

    #elif ATOMIC_LEVEL==0 //default
    (*(fmt_.m_data.processed_[sc.streamID()]))++;
    #endif
    eventCountForPathInit_[sc.streamID()]++;

    //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; 
    //fmt_.m_data.fastPathProcessedJ_ = totalEventsProcessed_.load(std::memory_order_relaxed);
  }

void evf::FastMonitoringService::postGlobalBeginRun

(

edm::GlobalContext const &

gc

)

Definition at line 390 of file FastMonitoringService.cc.

References isInitTransition_, macrostate_, and evf::FastMonitoringThread::sRunning.

  {
    macrostate_ = FastMonitoringThread::sRunning;
    isInitTransition_=false;
  }

void evf::FastMonitoringService::postGlobalEndLumi

(

edm::GlobalContext const &

gc

)

Definition at line 507 of file FastMonitoringService.cc.

References lastGlobalLumisClosed_, edm::LuminosityBlockID::luminosityBlock(), and edm::GlobalContext::luminosityBlockID().

Referenced by FastMonitoringService().

  {
    //mark closed lumis (still keep map entries until next one)
    lastGlobalLumisClosed_.push(gc.luminosityBlockID().luminosityBlock());
  }

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

Definition at line 632 of file FastMonitoringService.cc.

References evf::MicroStateService::mFwkOvhMod, microstate_, evf::MicroStateService::reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

  {
    //microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
    microstate_[sc.streamID().value()] = &reservedMicroStateNames[mFwkOvhMod];
  }

void evf::FastMonitoringService::postSourceEvent

(

edm::StreamID

sid

)

Definition at line 622 of file FastMonitoringService.cc.

References evf::MicroStateService::mFwkOvhSrc, microstate_, evf::MicroStateService::reservedMicroStateNames, and edm::StreamID::value().

Referenced by FastMonitoringService().

  {
    microstate_[sid.value()] = &reservedMicroStateNames[mFwkOvhSrc];
  }

void evf::FastMonitoringService::postStreamBeginLumi

(

edm::StreamContext const &

sc

)

Definition at line 526 of file FastMonitoringService.cc.

References microstate_, evf::MicroStateService::mIdle, evf::MicroStateService::reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

  {
    microstate_[sc.streamID().value()]=&reservedMicroStateNames[mIdle];
  }

void evf::FastMonitoringService::postStreamEndLumi

(

edm::StreamContext const &

sc

)

Definition at line 547 of file FastMonitoringService.cc.

References evf::MicroStateService::mFwkEoL, microstate_, evf::MicroStateService::reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

  {
    microstate_[sc.streamID().value()]=&reservedMicroStateNames[mFwkEoL];
  }

void evf::FastMonitoringService::preallocate

(

edm::service::SystemBounds const &

bounds

)

Definition at line 174 of file FastMonitoringService.cc.

References edm::service::SystemBounds::maxNumberOfStreams(), edm::service::SystemBounds::maxNumberOfThreads(), 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;
  }

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

Definition at line 183 of file FastMonitoringService.cc.

References collectedPathList_, evf::FastMonitoringService::Encoding::completeReservedWithDummies(), dowork(), emptyLumisectionMode_, encModule_, encPath_, eventCountForPathInit_, Exception, fastMicrostateDefPath_, fastName_, fastPath_, fastPathList_, filePerFwkStream_, firstEventId_, fmt_, i, evf::FastMonitoringThread::inCOUNT, inputLegendFileJson_, evf::FastMonitoringThread::MonitorData::inputstateBins_, isGlobalLumiTransition_, isInitTransition_, evf::FastMonitoringThread::jsonMonitor_, lastGlobalLumi_, LogDebug, lumiFromSource_, evf::FastMonitoringThread::m_data, macrostate_, evf::FastMonitoringThread::MonitorData::macrostateBins_, evf::FastMonitoringThread::MCOUNT, evf::MicroStateService::mCOUNT, microstate_, evf::FastMonitoringThread::MonitorData::microstateBins_, microstateDefPath_, ministate_, evf::FastMonitoringThread::MonitorData::ministateBins_, evf::MicroStateService::mInvalid, moduleLegendFile_, moduleLegendFileJson_, monInit_, nopath_, nStreams_, nThreads_, cppFunctionSkipper::operator, fed_dqm_sourceclient-live_cfg::path, pathLegendFile_, pathLegendFileJson_, evf::FastMonitoringThread::MonitorData::registerVariables(), evf::MicroStateService::reservedMicroStateNames, evf::FastMonitoringThread::resetFastMonitor(), runDirectory_, evf::FastMonitoringThread::sInit, evf::FastMonitoringThread::start(), streamCounterUpdating_, threadIDAvailable_, evf::FastMonitoringService::Encoding::updateReserved(), and workingDirectory_.

Referenced by FastMonitoringService().

  {

    // FIND RUN DIRECTORY
    // The run dir should be set via the configuration of EvFDaqDirector
    
    if (edm::Service<evf::EvFDaqDirector>().operator->()==nullptr)
    {
      throw cms::Exception("FastMonitoringService") << "EvFDaqDirector is not present";
    
    }
    emptyLumisectionMode_ = edm::Service<evf::EvFDaqDirector>()->emptyLumisectionMode();
    boost::filesystem::path runDirectory(edm::Service<evf::EvFDaqDirector>()->baseRunDir());
    workingDirectory_ = runDirectory_ = runDirectory;
    workingDirectory_ /= "mon";

    if ( !boost::filesystem::is_directory(workingDirectory_)) {
        LogDebug("FastMonitoringService") << "<MON> DIR NOT FOUND! Trying to create -: " << workingDirectory_.string() ;
        boost::filesystem::create_directories(workingDirectory_);
        if ( !boost::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";
    boost::filesystem::path fast = workingDirectory_;
    fast /= fastFileName.str();
    fastPath_ = fast.string();
    if (filePerFwkStream_)
      for (unsigned int i=0;i<nStreams_;i++) {
        std::ostringstream fastFileNameTid;
        fastFileNameTid << fastName_ << "_pid" << std::setfill('0') << std::setw(5) << getpid() << "_tid" << i << ".fast";
        boost::filesystem::path fastTid = workingDirectory_;
        fastTid /= fastFileNameTid.str();
        fastPathList_.push_back(fastTid.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 pathLegFile;
    std::ostringstream pathLegFileJson;
    pathLegFile << "pathlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
    pathLegendFile_  = (workingDirectory_/pathLegFile.str()).string();
    pathLegFileJson << "pathlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
    pathLegendFileJson_  = (workingDirectory_/pathLegFileJson.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_;
                              //<< encPath_.current_ + 1 << " " << encModule_.current_ + 1

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

    macrostate_=FastMonitoringThread::sInit;

    for(unsigned int i = 0; i < (mCOUNT); i++)
      encModule_.updateReserved(static_cast<const void*>(reservedMicroStateNames+i));
    encModule_.completeReservedWithDummies();

    for (unsigned int i=0;i<nStreams_;i++) {
       ministate_.push_back(&nopath_);
       microstate_.push_back(&reservedMicroStateNames[mInvalid]);

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

       //path (mini) state
       encPath_.emplace_back(0);
       encPath_[i].update(static_cast<const void*>(&nopath_));
       eventCountForPathInit_.push_back(0);
       firstEventId_.push_back(0);
       collectedPathList_.push_back(new std::atomic<bool>(0));

    }
    //for (unsigned int i=0;i<nThreads_;i++)
    //  threadMicrostate_.push_back(&reservedMicroStateNames[mInvalid]);

    //initial size until we detect number of bins
    fmt_.m_data.macrostateBins_=FastMonitoringThread::MCOUNT;
    fmt_.m_data.ministateBins_=0;
    fmt_.m_data.microstateBins_ = 0; 
    fmt_.m_data.inputstateBins_ = FastMonitoringThread::inCOUNT;
 
    lastGlobalLumi_=0; 
    isGlobalLumiTransition_=false;
    isInitTransition_=true;
    lumiFromSource_=0;

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

   //this definition needs: #include "tbb/compat/thread"
   //however this would results in TBB imeplementation replacing std::thread
   //(both supposedly call pthread_self())
   //number of threads created in process could be obtained from /proc,
   //assuming that all posix threads are true kernel threads capable of running in parallel

   //#if TBB_IMPLEMENT_CPP0X
   //threadIDAvailable_=true;
   //#endif

  }

void evf::FastMonitoringService::preEvent

(

edm::StreamContext const &

sc

)

Definition at line 586 of file FastMonitoringService.cc.

Referenced by FastMonitoringService().

  {
  }

void evf::FastMonitoringService::preGlobalBeginLumi

(

edm::GlobalContext const &

gc

)

Definition at line 396 of file FastMonitoringService.cc.

References accuSize_, avgLeadTime_, filesProcessedDuringLumi_, fmt_, isGlobalLumiTransition_, lastGlobalLumi_, lastGlobalLumisClosed_, CommonMethods::lock(), lockStatsDuringLumi_, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, evf::FastMonitoringThread::monlock_, and processedEventsPerLumi_.

Referenced by FastMonitoringService().

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

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

      lumiStartTime_[newLumi]=lumiStartTime;
      while (!lastGlobalLumisClosed_.empty()) {
          //wipe out old map entries as they aren't needed and slow down access
          unsigned int oldLumi = lastGlobalLumisClosed_.back();
          lastGlobalLumisClosed_.pop();
          lumiStartTime_.erase(oldLumi);
          avgLeadTime_.erase(oldLumi);
          filesProcessedDuringLumi_.erase(oldLumi);
          accuSize_.erase(oldLumi);
          lockStatsDuringLumi_.erase(oldLumi);
          processedEventsPerLumi_.erase(oldLumi);
      }
      lastGlobalLumi_= newLumi;
      isGlobalLumiTransition_=false;
  }

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

Definition at line 319 of file FastMonitoringService.cc.

References edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, exceptionInLS_, edm::ExternalSignal, fmt_, CommonMethods::lock(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), evf::FastMonitoringThread::monlock_, 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());
  }

void evf::FastMonitoringService::preGlobalEndLumi

(

edm::GlobalContext const &

gc

)

Definition at line 420 of file FastMonitoringService.cc.

References accuSize_, doSnapshot(), Exception, exception_detected_, exceptionInLS_, evf::FastMonitoringThread::MonitorData::fastThroughputJ_, filePerFwkStream_, fmt_, FedRawDataInputSource::getEventReport(), inputSource_, isGlobalLumiTransition_, evf::FastMonitoringThread::jsonMonitor_, CommonMethods::lock(), LogDebug, fjr2json::lumi, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, evf::FastMonitoringThread::m_data, evf::FastMonitoringThread::monlock_, fed_dqm_sourceclient-live_cfg::path, processedEventsPerLumi_, edm::shutdown_flag, slowName_, throughputFactor(), jsoncollector::IntJ::value(), jsoncollector::DoubleJ::value(), and workingDirectory_.

Referenced by FastMonitoringService().

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

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

      // Compute throughput
      timeval stt = lumiStartTime_[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];
      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_) {
            auto sourceReport  = inputSource_->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&output for this lumi
          
      // create file name for slow monitoring file
          if (filePerFwkStream_) {
        std::stringstream slowFileNameStem;
        slowFileNameStem << slowName_ << "_ls" << std::setfill('0') << std::setw(4)
            << lumi << "_pid" << std::setfill('0')
            << std::setw(5) << getpid();
        boost::filesystem::path slow = workingDirectory_;
        slow /= slowFileNameStem.str();
        fmt_.jsonMonitor_->outputFullJSONs(slow.string(),".jsn",lumi);
          }
          else {
        std::stringstream slowFileName;
        slowFileName << slowName_ << "_ls" << std::setfill('0') << std::setw(4)
            << lumi << "_pid" << std::setfill('0')
            << std::setw(5) << getpid() << ".jsn";
        boost::filesystem::path slow = workingDirectory_;
        slow /= slowFileName.str();
        fmt_.jsonMonitor_->outputFullJSON(slow.string(),lumi);//full global and stream merge and JSON write for this lumi
          }
      fmt_.jsonMonitor_->discardCollected(lumi);//we don't do further updates for this lumi

      isGlobalLumiTransition_=true;
  }

void evf::FastMonitoringService::preModuleBeginJob

(

edm::ModuleDescription const &

desc

)

Definition at line 353 of file FastMonitoringService.cc.

References encModule_, fmt_, CommonMethods::lock(), edm::ModuleDescription::moduleName(), evf::FastMonitoringThread::monlock_, nOutputModules_, evf::FastMonitoringService::Encoding::update(), and evf::FastMonitoringService::Encoding::updateReserved().

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() == "ShmStreamConsumer" || desc.moduleName() == "EvFOutputModule" ||
       desc.moduleName() == "EventStreamFileWriter" || desc.moduleName() == "PoolOutputModule") {
      encModule_.updateReserved((void*)&desc);
      nOutputModules_++;
    }
    else
      encModule_.update((void*)&desc);
  }

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

Definition at line 627 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

  {
    microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
  }

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

Definition at line 553 of file FastMonitoringService.cc.

References collectedPathList_, encPath_, edm::EventID::event(), eventCountForPathInit_, edm::StreamContext::eventID(), firstEventId_, fmt_, CommonMethods::lock(), evf::FastMonitoringThread::m_data, makePathLegendaJson(), ministate_, evf::FastMonitoringThread::MonitorData::ministateBins_, evf::FastMonitoringThread::monlock_, pathLegendFileJson_, pathLegendWritten_, edm::PathContext::pathName(), edm::StreamContext::streamID(), AlCaHLTBitMon_QueryRunRegistry::string, and unlikely.

Referenced by FastMonitoringService().

  {
    //make sure that all path names are retrieved before allowing ministate to change
    //hack: assume memory is synchronized after ~50 events seen by each stream
    if (unlikely(eventCountForPathInit_[sc.streamID()]<50) && false==collectedPathList_[sc.streamID()]->load(std::memory_order_acquire))
    {
      //protection between stream threads, as well as the service monitoring thread
      std::lock_guard<std::mutex> lock(fmt_.monlock_);

      if (firstEventId_[sc.streamID()]==0) 
    firstEventId_[sc.streamID()]=sc.eventID().event();
      if (sc.eventID().event()==firstEventId_[sc.streamID()])
      {
    encPath_[sc.streamID()].update((void*)&pc.pathName());
    return;
      }
      else {
    //finished collecting path names
    collectedPathList_[sc.streamID()]->store(true,std::memory_order_seq_cst);
    fmt_.m_data.ministateBins_=encPath_[sc.streamID()].vecsize();
    if (!pathLegendWritten_) {
      std::string pathLegendStrJson =  makePathLegendaJson();
      FileIO::writeStringToFile(pathLegendFileJson_, pathLegendStrJson);
      pathLegendWritten_=true;
    }
      }
    }
    else {
      ministate_[sc.streamID()] = &(pc.pathName());
    }
  }

void evf::FastMonitoringService::preSourceEarlyTermination

(

edm::TerminationOrigin

to

)

Definition at line 331 of file FastMonitoringService.cc.

References exception_detected_, edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, edm::ExternalSignal, fmt_, CommonMethods::lock(), evf::FastMonitoringThread::monlock_, 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; 
  }

void evf::FastMonitoringService::preSourceEvent

(

edm::StreamID

sid

)

Definition at line 617 of file FastMonitoringService.cc.

References microstate_, evf::MicroStateService::mInput, evf::MicroStateService::reservedMicroStateNames, and edm::StreamID::value().

Referenced by FastMonitoringService().

  {
    microstate_[sid.value()] = &reservedMicroStateNames[mInput];
  }

void evf::FastMonitoringService::preStreamBeginLumi

(

edm::StreamContext const &

sc

)

Definition at line 513 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), fmt_, CommonMethods::lock(), edm::EventID::luminosityBlock(), evf::FastMonitoringThread::m_data, evf::MicroStateService::mBoL, microstate_, ministate_, evf::FastMonitoringThread::monlock_, nopath_, evf::FastMonitoringThread::MonitorData::processed_, evf::MicroStateService::reservedMicroStateNames, edm::StreamContext::streamID(), evf::FastMonitoringThread::MonitorData::streamLumi_, and edm::StreamID::value().

Referenced by FastMonitoringService().

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

    //reset collected values for this stream
    *(fmt_.m_data.processed_[sid])=0;

    ministate_[sid]=&nopath_;
    microstate_[sid]=&reservedMicroStateNames[mBoL];
  }

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

Definition at line 307 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, exceptionInLS_, edm::ExternalSignal, fmt_, CommonMethods::lock(), edm::EventID::luminosityBlock(), evf::FastMonitoringThread::monlock_, 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());
  }

void evf::FastMonitoringService::preStreamEndLumi

(

edm::StreamContext const &

sc

)

Definition at line 531 of file FastMonitoringService.cc.

References doStreamEOLSnapshot(), edm::StreamContext::eventID(), fmt_, svgfig::load(), CommonMethods::lock(), edm::EventID::luminosityBlock(), evf::MicroStateService::mEoL, microstate_, ministate_, evf::FastMonitoringThread::monlock_, nopath_, evf::MicroStateService::reservedMicroStateNames, streamCounterUpdating_, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

  {
    unsigned int sid = sc.streamID().value();
    std::lock_guard<std::mutex> lock(fmt_.monlock_);

    #if ATOMIC_LEVEL>=2
    //spinlock to make sure we are not still updating event counter somewhere
    while (streamCounterUpdating_[sid]->load(std::memory_order_acquire)) {}
    #endif

    //update processed count to be complete at this time
    doStreamEOLSnapshot(sc.eventID().luminosityBlock(),sid);
    //reset this in case stream does not get notified of next lumi (we keep processed events only)
    ministate_[sid]=&nopath_;
    microstate_[sid]=&reservedMicroStateNames[mEoL];
  }

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

Definition at line 701 of file FastMonitoringService.cc.

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

Referenced by FedRawDataInputSource::readSupervisor().

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

  }

void evf::FastMonitoringService::setExceptionDetected

(

unsigned int

ls

)

Definition at line 342 of file FastMonitoringService.cc.

References exception_detected_, and exceptionInLS_.

Referenced by FedRawDataInputSource::getNextEvent().

                                                                  {
    if (!ls) exception_detected_=true;
    else exceptionInLS_.push_back(ls);
  }

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

inline

Definition at line 173 of file FastMonitoringService.h.

References inputSource_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

{inputSource_=inputSource;}

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

inline

Definition at line 174 of file FastMonitoringService.h.

References inputState_.

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

{inputState_=inputState;}

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

inline

Definition at line 175 of file FastMonitoringService.h.

References inputSupervisorState_.

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

{inputSupervisorState_=inputState;}

void evf::FastMonitoringService::setMicroState ( MicroStateService::Microstate  m )

virtual

Implements evf::MicroStateService.

Definition at line 642 of file FastMonitoringService.cc.

References i, microstate_, nStreams_, and evf::MicroStateService::reservedMicroStateNames.

  {
    for (unsigned int i=0;i<nStreams_;i++)
      microstate_[i] = &reservedMicroStateNames[m];
  }

void evf::FastMonitoringService::setMicroState ( edm::StreamID  sid, MicroStateService::Microstate  m  )

virtual

Implements evf::MicroStateService.

Definition at line 649 of file FastMonitoringService.cc.

References visualization-live-secondInstance_cfg::m, microstate_, and evf::MicroStateService::reservedMicroStateNames.

  {
    microstate_[sid] = &reservedMicroStateNames[m];
  }

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

inline

Definition at line 166 of file FastMonitoringService.h.

References emptyLumisectionMode_, getAbortFlagForLumi(), getEventsProcessedForLumi(), and proc.

Referenced by DQMFileSaver::globalEndLuminosityBlock(), HLTriggerJSONMonitoring::globalEndLuminosityBlockSummary(), L1TriggerJSONMonitoring::globalEndLuminosityBlockSummary(), and TriggerJSONMonitoring::globalEndLuminosityBlockSummary().

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

void evf::FastMonitoringService::startedLookingForFile

(

)

Definition at line 667 of file FastMonitoringService.cc.

References fileLookStart_.

Referenced by FedRawDataInputSource::readSupervisor().

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

void evf::FastMonitoringService::stoppedLookingForFile

(

unsigned int

lumi

)

Definition at line 675 of file FastMonitoringService.cc.

References avgLeadTime_, fileLookStart_, fileLookStop_, fmt_, i, leadTimes_, CommonMethods::lock(), fjr2json::lumi, lumiFromSource_, and evf::FastMonitoringThread::monlock_.

Referenced by FedRawDataInputSource::readSupervisor().

                                                                     {
      gettimeofday(&fileLookStop_, 0);
      /*
     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

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

private

Definition at line 270 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), preGlobalBeginLumi(), and preGlobalEndLumi().

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

private

Definition at line 267 of file FastMonitoringService.h.

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

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

private

Definition at line 282 of file FastMonitoringService.h.

Referenced by preBeginJob(), and prePathEvent().

bool evf::FastMonitoringService::emptyLumisectionMode_ = false

private

Definition at line 303 of file FastMonitoringService.h.

Referenced by preBeginJob(), and shouldWriteFiles().

Encoding evf::FastMonitoringService::encModule_

private

Definition at line 231 of file FastMonitoringService.h.

Referenced by doSnapshot(), dowork(), makeModuleLegendaJson(), postBeginJob(), preBeginJob(), and preModuleBeginJob().

std::vector<Encoding> evf::FastMonitoringService::encPath_

private

Definition at line 232 of file FastMonitoringService.h.

Referenced by doSnapshot(), dowork(), makePathLegendaJson(), preBeginJob(), and prePathEvent().

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

private

Definition at line 283 of file FastMonitoringService.h.

Referenced by postEvent(), preBeginJob(), and prePathEvent().

bool evf::FastMonitoringService::exception_detected_ = false

private

Definition at line 301 of file FastMonitoringService.h.

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

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

private

Definition at line 302 of file FastMonitoringService.h.

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

std::string evf::FastMonitoringService::fastMicrostateDefPath_

private

Definition at line 242 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

unsigned int evf::FastMonitoringService::fastMonIntervals_

private

Definition at line 240 of file FastMonitoringService.h.

Referenced by dowork().

std::string evf::FastMonitoringService::fastName_

private

Definition at line 243 of file FastMonitoringService.h.

Referenced by preBeginJob().

std::string evf::FastMonitoringService::fastPath_

private

Definition at line 243 of file FastMonitoringService.h.

Referenced by dowork(), and preBeginJob().

std::vector<std::string> evf::FastMonitoringService::fastPathList_

private

Definition at line 304 of file FastMonitoringService.h.

Referenced by dowork(), and preBeginJob().

timeval evf::FastMonitoringService::fileLookStart_

private

Definition at line 249 of file FastMonitoringService.h.

Referenced by startedLookingForFile(), and stoppedLookingForFile().

timeval evf::FastMonitoringService::fileLookStop_

private

Definition at line 249 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

bool evf::FastMonitoringService::filePerFwkStream_

private

Definition at line 244 of file FastMonitoringService.h.

Referenced by dowork(), preBeginJob(), and preGlobalEndLumi().

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

private

Definition at line 268 of file FastMonitoringService.h.

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

std::vector<unsigned long> evf::FastMonitoringService::firstEventId_

private

Definition at line 281 of file FastMonitoringService.h.

Referenced by preBeginJob(), and prePathEvent().

FastMonitoringThread evf::FastMonitoringService::fmt_

private

Definition at line 230 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), doSnapshot(), doStreamEOLSnapshot(), dowork(), getAbortFlagForLumi(), getEventsProcessedForLumi(), postBeginJob(), postEndJob(), postEvent(), preBeginJob(), preGlobalBeginLumi(), preGlobalEarlyTermination(), preGlobalEndLumi(), preModuleBeginJob(), prePathEvent(), preSourceEarlyTermination(), preStreamBeginLumi(), preStreamEarlyTermination(), preStreamEndLumi(), reportLockWait(), and stoppedLookingForFile().

std::string evf::FastMonitoringService::inputLegendFileJson_

private

Definition at line 296 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

FedRawDataInputSource* evf::FastMonitoringService::inputSource_ = 0

private

Definition at line 233 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

FastMonitoringThread::InputState evf::FastMonitoringService::inputState_

private

Definition at line 234 of file FastMonitoringService.h.

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

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

static

Initial value:

= 
    {"Ignore","Init","WaitInput","NewLumi","NewLumiBusyEndingLS","NewLumiIdleEndingLS","RunEnd","ProcessingFile","WaitChunk","ChunkReceived",
     "ChecksumEvent","CachedEvent","ReadEvent","ReadCleanup","NoRequest","NoRequestWithIdleThreads",
     "NoRequestWithGlobalEoL","NoRequestWithEoLThreads",
     "SupFileLimit", "SupWaitFreeChunk","SupWaitFreeChunkCopying", "SupWaitFreeThread","SupWaitFreeThreadCopying",
     "SupBusy", "SupLockPolling","SupLockPollingCopying",
     "SupNoFile", "SupNewFile", "SupNewFileWaitThreadCopying", "SupNewFileWaitThread",
     "SupNewFileWaitChunkCopying", "SupNewFileWaitChunk",
     "WaitInput_fileLimit","WaitInput_waitFreeChunk","WaitInput_waitFreeChunkCopying","WaitInput_waitFreeThread","WaitInput_waitFreeThreadCopying",
     "WaitInput_busy","WaitInput_lockPolling","WaitInput_lockPollingCopying","WaitInput_runEnd",
     "WaitInput_noFile","WaitInput_newFile","WaitInput_newFileWaitThreadCopying","WaitInput_newFileWaitThread",
     "WaitInput_newFileWaitChunkCopying","WaitInput_newFileWaitChunk",
     "WaitChunk_fileLimit","WaitChunk_waitFreeChunk","WaitChunk_waitFreeChunkCopying","WaitChunk_waitFreeThread","WaitChunk_waitFreeThreadCopying",
     "WaitChunk_busy","WaitChunk_lockPolling","WaitChunk_lockPollingCopying","WaitChunk_runEnd",
     "WaitChunk_noFile","WaitChunk_newFile","WaitChunk_newFileWaitThreadCopying","WaitChunk_newFileWaitThread",
     "WaitChunk_newFileWaitChunkCopying","WaitChunk_newFileWaitChunk"
    }

Definition at line 115 of file FastMonitoringService.h.

Referenced by dowork(), and makeInputLegendaJson().

FastMonitoringThread::InputState evf::FastMonitoringService::inputSupervisorState_

private

Definition at line 235 of file FastMonitoringService.h.

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

bool evf::FastMonitoringService::isGlobalLumiTransition_

private

Definition at line 253 of file FastMonitoringService.h.

Referenced by doSnapshot(), preBeginJob(), preGlobalBeginLumi(), and preGlobalEndLumi().

bool evf::FastMonitoringService::isInitTransition_

private

Definition at line 254 of file FastMonitoringService.h.

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

unsigned int evf::FastMonitoringService::lastGlobalLumi_

private

Definition at line 251 of file FastMonitoringService.h.

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

std::queue<unsigned int> evf::FastMonitoringService::lastGlobalLumisClosed_

private

Definition at line 252 of file FastMonitoringService.h.

Referenced by postGlobalEndLumi(), and preGlobalBeginLumi().

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

private

Definition at line 271 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

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

private

Definition at line 272 of file FastMonitoringService.h.

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

unsigned int evf::FastMonitoringService::lumiFromSource_

private

Definition at line 255 of file FastMonitoringService.h.

Referenced by preBeginJob(), and stoppedLookingForFile().

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

private

Definition at line 248 of file FastMonitoringService.h.

Referenced by preGlobalBeginLumi(), and preGlobalEndLumi().

FastMonitoringThread::Macrostate evf::FastMonitoringService::macrostate_

private

Definition at line 258 of file FastMonitoringService.h.

Referenced by doSnapshot(), jobFailure(), postBeginJob(), postEndJob(), postGlobalBeginRun(), and preBeginJob().

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

static

Initial value:

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

Definition at line 114 of file FastMonitoringService.h.

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

private

Definition at line 262 of file FastMonitoringService.h.

Referenced by doSnapshot(), dowork(), postEvent(), postModuleEvent(), postSourceEvent(), postStreamBeginLumi(), postStreamEndLumi(), preBeginJob(), preModuleEvent(), preSourceEvent(), preStreamBeginLumi(), preStreamEndLumi(), and setMicroState().

std::string evf::FastMonitoringService::microstateDefPath_

private

Definition at line 242 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

std::vector<const void*> evf::FastMonitoringService::ministate_

private

Definition at line 261 of file FastMonitoringService.h.

Referenced by doSnapshot(), dowork(), postEvent(), preBeginJob(), prePathEvent(), preStreamBeginLumi(), and preStreamEndLumi().

std::string evf::FastMonitoringService::moduleLegendFile_

private

Definition at line 292 of file FastMonitoringService.h.

Referenced by preBeginJob().

std::string evf::FastMonitoringService::moduleLegendFileJson_

private

Definition at line 293 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

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

private

Definition at line 300 of file FastMonitoringService.h.

Referenced by dowork(), and preBeginJob().

const std::string evf::FastMonitoringService::nopath_ = "NoPath"

static

Definition at line 118 of file FastMonitoringService.h.

Referenced by postEvent(), preBeginJob(), preStreamBeginLumi(), and preStreamEndLumi().

unsigned int evf::FastMonitoringService::nOutputModules_ =0

private

Definition at line 298 of file FastMonitoringService.h.

Referenced by makeModuleLegendaJson(), and preModuleBeginJob().

unsigned int evf::FastMonitoringService::nStreams_

private

Definition at line 237 of file FastMonitoringService.h.

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

unsigned int evf::FastMonitoringService::nThreads_

private

Definition at line 238 of file FastMonitoringService.h.

Referenced by preallocate(), and preBeginJob().

std::string evf::FastMonitoringService::pathLegendFile_

private

Definition at line 294 of file FastMonitoringService.h.

Referenced by preBeginJob().

std::string evf::FastMonitoringService::pathLegendFileJson_

private

Definition at line 295 of file FastMonitoringService.h.

Referenced by preBeginJob(), and prePathEvent().

bool evf::FastMonitoringService::pathLegendWritten_ = false

private

Definition at line 297 of file FastMonitoringService.h.

Referenced by prePathEvent().

std::vector<bool> evf::FastMonitoringService::pathNamesReady_

private

Definition at line 284 of file FastMonitoringService.h.

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

private

Definition at line 275 of file FastMonitoringService.h.

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

boost::filesystem::path evf::FastMonitoringService::runDirectory_

private

Definition at line 286 of file FastMonitoringService.h.

Referenced by getRunDirName(), and preBeginJob().

int evf::FastMonitoringService::sleepTime_

private

Definition at line 239 of file FastMonitoringService.h.

Referenced by dowork().

std::string evf::FastMonitoringService::slowName_

private

Definition at line 243 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi().

unsigned int evf::FastMonitoringService::snapCounter_ = 0

private

Definition at line 241 of file FastMonitoringService.h.

Referenced by dowork().

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

private

Definition at line 279 of file FastMonitoringService.h.

Referenced by postEvent(), preBeginJob(), and preStreamEndLumi().

bool evf::FastMonitoringService::threadIDAvailable_ = false

private

Definition at line 288 of file FastMonitoringService.h.

Referenced by preBeginJob().

std::vector<const void*> evf::FastMonitoringService::threadMicrostate_

private

Definition at line 263 of file FastMonitoringService.h.

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

private

Definition at line 290 of file FastMonitoringService.h.

Referenced by postEvent().

boost::filesystem::path evf::FastMonitoringService::workingDirectory_

private

Definition at line 286 of file FastMonitoringService.h.

Referenced by preBeginJob(), and preGlobalEndLumi().