CMS 3D CMS Logo

List of all members | Public Member Functions | Static Public Member Functions | Static Public Attributes | Private Member Functions | Private Attributes
evf::FastMonitoringService Class Reference

#include <FastMonitoringService.h>

Inheritance diagram for evf::FastMonitoringService:
evf::MicroStateService

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 ()
 
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 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 setInState (FastMonState::InputState inputState)
 
void setInStateSup (FastMonState::InputState inputState)
 
void setMicroState (FastMonState::Microstate)
 
void setMicroState (edm::StreamID, FastMonState::Microstate)
 
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 Public Attributes

static const std::string inputStateNames [FastMonState::inCOUNT]
 
static const std::string macroStateNames [FastMonState::MCOUNT]
 
static const std::string nopath_ = "NoPath"
 
static const edm::ModuleDescription reservedMicroStateNames [FastMonState::mCOUNT]
 

Private Member Functions

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

Private Attributes

std::map< unsigned int, unsigned long > accuSize_
 
std::map< unsigned int, double > avgLeadTime_
 
std::vector< std::atomic< bool > * > collectedPathList_
 
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::shared_ptr< FastMonitoringThreadfmt_
 
std::atomic< bool > has_data_exception_ = false
 
std::atomic< bool > has_source_exception_ = false
 
std::string inputLegendFileJson_
 
FedRawDataInputSourceinputSource_ = nullptr
 
std::atomic< FastMonState::InputStateinputState_ {FastMonState::InputState::inInit}
 
std::atomic< FastMonState::InputStateinputSupervisorState_ {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::string microstateDefPath_
 
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_
 
std::vector< bool > pathNamesReady_
 
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
 
std::filesystem::path runDirectory_
 
int sleepTime_
 
std::string slowName_
 
unsigned int snapCounter_ = 0
 
std::vector< std::atomic< bool > * > streamCounterUpdating_
 
bool threadIDAvailable_ = false
 
std::atomic< unsigned long > totalEventsProcessed_
 
std::filesystem::path workingDirectory_
 

Detailed Description

Definition at line 155 of file FastMonitoringService.h.

Constructor & Destructor Documentation

◆ FastMonitoringService()

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

Definition at line 125 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(), prePathEvent(), 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::watchPrePathEvent(), edm::ActivityRegistry::watchPreSourceEarlyTermination(), edm::ActivityRegistry::watchPreSourceEvent(), edm::ActivityRegistry::watchPreStreamBeginLumi(), edm::ActivityRegistry::watchPreStreamEarlyTermination(), and edm::ActivityRegistry::watchPreStreamEndLumi().

126  : MicroStateService(iPS, reg),
127  fmt_(new FastMonitoringThread()),
128  nStreams_(0) //until initialized
129  ,
130  sleepTime_(iPS.getUntrackedParameter<int>("sleepTime", 1)),
131  fastMonIntervals_(iPS.getUntrackedParameter<unsigned int>("fastMonIntervals", 2)),
132  fastName_("fastmoni"),
133  slowName_("slowmoni"),
134  filePerFwkStream_(iPS.getUntrackedParameter<bool>("filePerFwkStream", false)),
136  reg.watchPreallocate(this, &FastMonitoringService::preallocate); //receiving information on number of threads
138 
143 
147 
152 
154 
155  reg.watchPreEvent(this, &FastMonitoringService::preEvent); //stream
157 
158  reg.watchPreSourceEvent(this, &FastMonitoringService::preSourceEvent); //source (with streamID of requestor)
160 
163 
166 
170 
171  //find microstate definition path (required by the module)
172  struct stat statbuf;
173  std::string microstateBaseSuffix = "src/EventFilter/Utilities/plugins/microstatedef.jsd";
174  std::string microstatePath = std::string(std::getenv("CMSSW_BASE")) + "/" + microstateBaseSuffix;
175  if (stat(microstatePath.c_str(), &statbuf)) {
176  microstatePath = std::string(std::getenv("CMSSW_RELEASE_BASE")) + "/" + microstateBaseSuffix;
177  if (stat(microstatePath.c_str(), &statbuf)) {
178  microstatePath = microstateBaseSuffix;
179  if (stat(microstatePath.c_str(), &statbuf))
180  throw cms::Exception("FastMonitoringService") << "microstate definition file not found";
181  }
182  }
183  fastMicrostateDefPath_ = microstateDefPath_ = microstatePath;
184  }
void prePathEvent(edm::StreamContext const &, edm::PathContext const &)
void watchPreStreamEarlyTermination(PreStreamEarlyTermination::slot_type const &iSlot)
void watchPreEvent(PreEvent::slot_type const &iSlot)
void postModuleEventAcquire(edm::StreamContext const &, edm::ModuleCallingContext const &)
void watchPrePathEvent(PrePathEvent::slot_type const &iSlot)
void watchPreallocate(Preallocate::slot_type const &iSlot)
void watchPreModuleEventAcquire(PreModuleEventAcquire::slot_type const &iSlot)
void watchPostEndJob(PostEndJob::slot_type const &iSlot)
void preallocate(edm::service::SystemBounds const &)
void preGlobalBeginLumi(edm::GlobalContext const &)
void watchPreModuleEvent(PreModuleEvent::slot_type const &iSlot)
void postGlobalEndLumi(edm::GlobalContext const &)
void postEvent(edm::StreamContext const &)
void watchPreGlobalEndLumi(PreGlobalEndLumi::slot_type const &iSlot)
void watchPostEvent(PostEvent::slot_type const &iSlot)
void watchPostStreamEndLumi(PostStreamEndLumi::slot_type const &iSlot)
void watchPreGlobalBeginLumi(PreGlobalBeginLumi::slot_type const &iSlot)
void watchPostModuleEvent(PostModuleEvent::slot_type const &iSlot)
void watchPostSourceEvent(PostSourceEvent::slot_type const &iSlot)
void preGlobalEndLumi(edm::GlobalContext const &)
void watchPreStreamEndLumi(PreStreamEndLumi::slot_type const &iSlot)
void preGlobalEarlyTermination(edm::GlobalContext const &, edm::TerminationOrigin)
void watchPreSourceEarlyTermination(PreSourceEarlyTermination::slot_type const &iSlot)
void watchJobFailure(JobFailure::slot_type const &iSlot)
convenience function for attaching to signal
T getUntrackedParameter(std::string const &, T const &) const
void preModuleBeginJob(edm::ModuleDescription const &)
MicroStateService(const edm::ParameterSet &, edm::ActivityRegistry &)
void preStreamEndLumi(edm::StreamContext const &)
void watchPostStreamBeginLumi(PostStreamBeginLumi::slot_type const &iSlot)
void preStreamEarlyTermination(edm::StreamContext const &, edm::TerminationOrigin)
void watchPreGlobalEarlyTermination(PreGlobalEarlyTermination::slot_type const &iSlot)
void watchPostModuleEventAcquire(PostModuleEventAcquire::slot_type const &iSlot)
void watchPostGlobalEndLumi(PostGlobalEndLumi::slot_type const &iSlot)
void watchPreModuleBeginJob(PreModuleBeginJob::slot_type const &iSlot)
void preModuleEventAcquire(edm::StreamContext const &, edm::ModuleCallingContext const &)
void postModuleEvent(edm::StreamContext const &, edm::ModuleCallingContext const &)
void postStreamBeginLumi(edm::StreamContext const &)
void postStreamEndLumi(edm::StreamContext const &)
void preStreamBeginLumi(edm::StreamContext const &)
std::atomic< unsigned long > totalEventsProcessed_
void watchPreStreamBeginLumi(PreStreamBeginLumi::slot_type const &iSlot)
void preBeginJob(edm::PathsAndConsumesOfModulesBase const &, edm::ProcessContext const &pc)
void watchPreBeginJob(PreBeginJob::slot_type const &iSlot)
convenience function for attaching to signal
void preEvent(edm::StreamContext const &)
void preSourceEarlyTermination(edm::TerminationOrigin)
void watchPreSourceEvent(PreSourceEvent::slot_type const &iSlot)
std::shared_ptr< FastMonitoringThread > fmt_
void preModuleEvent(edm::StreamContext const &, edm::ModuleCallingContext const &)
void watchPostBeginJob(PostBeginJob::slot_type const &iSlot)
convenience function for attaching to signal

◆ ~FastMonitoringService()

evf::FastMonitoringService::~FastMonitoringService ( )
override

Definition at line 186 of file FastMonitoringService.cc.

186 {}

Member Function Documentation

◆ accumulateFileSize()

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

Definition at line 692 of file FastMonitoringService.cc.

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

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

692  {
693  std::lock_guard<std::mutex> lock(fmt_->monlock_);
694 
695  if (accuSize_.find(lumi) == accuSize_.end())
696  accuSize_[lumi] = fileSize;
697  else
698  accuSize_[lumi] += fileSize;
699 
702  else
704  }
std::map< unsigned int, unsigned long > accuSize_
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::shared_ptr< FastMonitoringThread > fmt_

◆ doSnapshot()

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

Definition at line 841 of file FastMonitoringService.cc.

References avgLeadTime_, filesProcessedDuringLumi_, fmt_, mps_fire::i, caHitNtupletGeneratorKernels::if(), 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(), evf::FastMonState::mEoL, evf::FastMonState::mFwkEoL, evf::FastMonState::mIdle, nStreams_, and reservedMicroStateNames.

Referenced by preGlobalEndLumi(), and snapshotRunner().

841  {
842  // update macrostate
843  fmt_->m_data.fastMacrostateJ_ = fmt_->m_data.macrostate_;
844 
845  std::vector<const void*> microstateCopy(fmt_->m_data.microstate_.begin(), fmt_->m_data.microstate_.end());
846  std::vector<unsigned char> microstateAcqCopy(fmt_->m_data.microstateAcqFlag_.begin(),
847  fmt_->m_data.microstateAcqFlag_.end());
848 
849  if (!isInitTransition_) {
850  auto itd = avgLeadTime_.find(ls);
851  if (itd != avgLeadTime_.end())
852  fmt_->m_data.fastAvgLeadTimeJ_ = itd->second;
853  else
854  fmt_->m_data.fastAvgLeadTimeJ_ = 0.;
855 
856  auto iti = filesProcessedDuringLumi_.find(ls);
857  if (iti != filesProcessedDuringLumi_.end())
858  fmt_->m_data.fastFilesProcessedJ_ = iti->second;
859  else
860  fmt_->m_data.fastFilesProcessedJ_ = 0;
861 
862  auto itrd = lockStatsDuringLumi_.find(ls);
863  if (itrd != lockStatsDuringLumi_.end()) {
864  fmt_->m_data.fastLockWaitJ_ = itrd->second.first;
865  fmt_->m_data.fastLockCountJ_ = itrd->second.second;
866  } else {
867  fmt_->m_data.fastLockWaitJ_ = 0.;
868  fmt_->m_data.fastLockCountJ_ = 0.;
869  }
870  }
871 
872  for (unsigned int i = 0; i < nStreams_; i++) {
873  fmt_->m_data.ministateEncoded_[i] = fmt_->m_data.encPath_[i].encodeString(fmt_->m_data.ministate_[i]);
874  if (microstateAcqCopy[i])
875  fmt_->m_data.microstateEncoded_[i] =
876  fmt_->m_data.microstateBins_ + fmt_->m_data.encModule_.encode(microstateCopy[i]);
877  else
878  fmt_->m_data.microstateEncoded_[i] = fmt_->m_data.encModule_.encode(microstateCopy[i]);
879  }
880 
881  bool inputStatePerThread = false;
882 
884  switch (inputSupervisorState_) {
886  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_fileLimit;
887  break;
889  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeChunk;
890  break;
892  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeChunkCopying;
893  break;
895  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeThread;
896  break;
898  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeThreadCopying;
899  break;
901  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_busy;
902  break;
904  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_lockPolling;
905  break;
907  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_lockPollingCopying;
908  break;
910  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_runEnd;
911  break;
913  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_noFile;
914  break;
916  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFile;
917  break;
920  break;
922  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitThread;
923  break;
926  break;
928  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitChunk;
929  break;
930  default:
931  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput;
932  }
933  } else if (inputState_ == FastMonState::inWaitChunk) {
934  switch (inputSupervisorState_) {
936  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_fileLimit;
937  break;
939  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeChunk;
940  break;
942  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeChunkCopying;
943  break;
945  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeThread;
946  break;
948  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeThreadCopying;
949  break;
951  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_busy;
952  break;
954  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_lockPolling;
955  break;
957  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_lockPollingCopying;
958  break;
960  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_runEnd;
961  break;
963  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_noFile;
964  break;
966  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFile;
967  break;
970  break;
972  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitThread;
973  break;
976  break;
978  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitChunk;
979  break;
980  default:
981  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk;
982  }
983  } else if (inputState_ == FastMonState::inNoRequest) {
984  inputStatePerThread = true;
985  for (unsigned int i = 0; i < nStreams_; i++) {
986  if (microstateCopy[i] == &reservedMicroStateNames[FastMonState::mIdle])
987  fmt_->m_data.inputState_[i] = FastMonState::inNoRequestWithIdleThreads;
988  else if (microstateCopy[i] == &reservedMicroStateNames[FastMonState::mEoL] ||
989  microstateCopy[i] == &reservedMicroStateNames[FastMonState::mFwkEoL])
990  fmt_->m_data.inputState_[i] = FastMonState::inNoRequestWithEoLThreads;
991  else
992  fmt_->m_data.inputState_[i] = FastMonState::inNoRequest;
993  }
994  } else if (inputState_ == FastMonState::inNewLumi) {
995  inputStatePerThread = true;
996  for (unsigned int i = 0; i < nStreams_; i++) {
997  if (microstateCopy[i] == &reservedMicroStateNames[FastMonState::mEoL] ||
998  microstateCopy[i] == &reservedMicroStateNames[FastMonState::mFwkEoL])
999  fmt_->m_data.inputState_[i] = FastMonState::inNewLumi;
1000  }
1002  //apply directly throttled state from supervisor
1003  fmt_->m_data.inputState_[0] = inputSupervisorState_;
1004  } else
1005  fmt_->m_data.inputState_[0] = inputState_;
1006 
1007  //this is same for all streams
1008  if (!inputStatePerThread)
1009  for (unsigned int i = 1; i < nStreams_; i++)
1010  fmt_->m_data.inputState_[i] = fmt_->m_data.inputState_[0];
1011 
1012  if (isGlobalEOL) { //only update global variables
1013  fmt_->jsonMonitor_->snapGlobal(ls);
1014  } else
1015  fmt_->jsonMonitor_->snap(ls);
1016  }
std::atomic< FastMonState::InputState > inputState_
std::atomic< bool > isInitTransition_
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
std::map< unsigned int, double > avgLeadTime_
std::atomic< FastMonState::InputState > inputSupervisorState_
def ls(path, rec=False)
Definition: eostools.py:349
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
std::shared_ptr< FastMonitoringThread > fmt_

◆ exceptionDetected()

bool evf::FastMonitoringService::exceptionDetected ( ) const

Definition at line 431 of file FastMonitoringService.cc.

References has_data_exception_, and has_source_exception_.

Referenced by FedRawDataInputSource::~FedRawDataInputSource().

431  {
432  return has_source_exception_.load() || has_data_exception_.load();
433  }
std::atomic< bool > has_data_exception_
std::atomic< bool > has_source_exception_

◆ fillDescriptions()

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

Definition at line 188 of file FastMonitoringService.cc.

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

188  {
190  desc.setComment("Service for File-based DAQ monitoring and event accounting");
191  desc.addUntracked<int>("sleepTime", 1)->setComment("Sleep time of the monitoring thread");
192  desc.addUntracked<unsigned int>("fastMonIntervals", 2)
193  ->setComment("Modulo of sleepTime intervals on which fastmon file is written out");
194  desc.addUntracked<bool>("filePerFwkStream", false)
195  ->setComment("Switches on monitoring output per framework stream");
196  desc.setAllowAnything();
197  descriptions.add("FastMonitoringService", desc);
198  }
void add(std::string const &label, ParameterSetDescription const &psetDescription)

◆ getAbortFlagForLumi()

bool evf::FastMonitoringService::getAbortFlagForLumi ( unsigned int  lumi)

Definition at line 766 of file FastMonitoringService.cc.

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

Referenced by shouldWriteFiles().

766  {
767  std::lock_guard<std::mutex> lock(fmt_->monlock_);
768 
769  auto it = processedEventsPerLumi_.find(lumi);
770  if (it != processedEventsPerLumi_.end()) {
771  unsigned int abortFlag = it->second.second;
772  return abortFlag;
773  } else {
774  throw cms::Exception("FastMonitoringService")
775  << "output module wants already deleted (or never reported by SOURCE) lumisection status for LUMI -: "
776  << lumi;
777  return false;
778  }
779  }
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
std::shared_ptr< FastMonitoringThread > fmt_

◆ getEventsProcessedForLumi()

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

Definition at line 748 of file FastMonitoringService.cc.

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

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

748  {
749  std::lock_guard<std::mutex> lock(fmt_->monlock_);
750 
751  auto it = processedEventsPerLumi_.find(lumi);
752  if (it != processedEventsPerLumi_.end()) {
753  unsigned int proc = it->second.first;
754  if (abortFlag)
755  *abortFlag = it->second.second;
756  return proc;
757  } else {
758  throw cms::Exception("FastMonitoringService")
759  << "output module wants already deleted (or never reported by SOURCE) lumisection event count for LUMI -: "
760  << lumi;
761  return 0;
762  }
763  }
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
std::shared_ptr< FastMonitoringThread > fmt_

◆ getRunDirName()

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

Definition at line 220 of file FastMonitoringService.h.

References runDirectory_.

220 { return runDirectory_.stem().string(); }
std::filesystem::path runDirectory_

◆ isExceptionOnData()

bool evf::FastMonitoringService::isExceptionOnData ( unsigned int  ls)

Definition at line 435 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(), and FedRawDataInputSource::~FedRawDataInputSource().

435  {
436  if (!has_data_exception_.load())
437  return false;
438  if (has_source_exception_.load())
439  return true;
440  std::lock_guard<std::mutex> lock(fmt_->monlock_);
441  for (auto ex : exceptionInLS_) {
442  if (ls == ex)
443  return true;
444  }
445  return false;
446  }
std::atomic< bool > has_data_exception_
def ls(path, rec=False)
Definition: eostools.py:349
std::atomic< bool > has_source_exception_
std::shared_ptr< FastMonitoringThread > fmt_
std::vector< unsigned int > exceptionInLS_

◆ jobFailure()

void evf::FastMonitoringService::jobFailure ( )

Definition at line 448 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

448 { fmt_->m_data.macrostate_ = FastMonState::sError; }
std::shared_ptr< FastMonitoringThread > fmt_

◆ makeInputLegendaJson()

std::string evf::FastMonitoringService::makeInputLegendaJson ( )

Definition at line 233 of file FastMonitoringService.cc.

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

Referenced by postBeginJob().

233  {
234  Json::Value legendaVector(Json::arrayValue);
235  for (int i = 0; i < FastMonState::inCOUNT; i++)
236  legendaVector.append(Json::Value(inputStateNames[i]));
237  Json::Value moduleLegend;
238  moduleLegend["names"] = legendaVector;
240  return writer.write(moduleLegend);
241  }
static const std::string inputStateNames[FastMonState::inCOUNT]
Represents a JSON value.
Definition: value.h:99
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:63
array value (ordered list)
Definition: value.h:30

◆ makeModuleLegendaJson()

std::string evf::FastMonitoringService::makeModuleLegendaJson ( )

Definition at line 212 of file FastMonitoringService.cc.

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

Referenced by postBeginJob().

212  {
213  Json::Value legendaVector(Json::arrayValue);
214  for (int i = 0; i < fmt_->m_data.encModule_.current_; i++)
215  legendaVector.append(
216  Json::Value((static_cast<const edm::ModuleDescription*>(fmt_->m_data.encModule_.decode(i)))->moduleLabel()));
217  //duplicate modules adding a list for acquire states (not all modules actually have it)
218  for (int i = 0; i < fmt_->m_data.encModule_.current_; i++)
219  legendaVector.append(Json::Value(
220  (static_cast<const edm::ModuleDescription*>(fmt_->m_data.encModule_.decode(i)))->moduleLabel() + "__ACQ"));
221  Json::Value valReserved(nReservedModules);
222  Json::Value valSpecial(nSpecialModules);
223  Json::Value valOutputModules(nOutputModules_);
224  Json::Value moduleLegend;
225  moduleLegend["names"] = legendaVector;
226  moduleLegend["reserved"] = valReserved;
227  moduleLegend["special"] = valSpecial;
228  moduleLegend["output"] = valOutputModules;
230  return writer.write(moduleLegend);
231  }
Represents a JSON value.
Definition: value.h:99
constexpr int nSpecialModules
constexpr int nReservedModules
std::shared_ptr< FastMonitoringThread > fmt_
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:63
array value (ordered list)
Definition: value.h:30

◆ makePathLegendaJson()

std::string evf::FastMonitoringService::makePathLegendaJson ( )

Definition at line 200 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, fmt_, mps_fire::i, evf::nReservedPaths, and convertToRaw::writer.

Referenced by postBeginJob().

200  {
201  Json::Value legendaVector(Json::arrayValue);
202  for (int i = 0; i < fmt_->m_data.encPath_[0].current_; i++)
203  legendaVector.append(Json::Value(*(static_cast<const std::string*>(fmt_->m_data.encPath_[0].decode(i)))));
204  Json::Value valReserved(nReservedPaths);
205  Json::Value pathLegend;
206  pathLegend["names"] = legendaVector;
207  pathLegend["reserved"] = valReserved;
209  return writer.write(pathLegend);
210  }
Represents a JSON value.
Definition: value.h:99
constexpr int nReservedPaths
std::shared_ptr< FastMonitoringThread > fmt_
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:63
array value (ordered list)
Definition: value.h:30

◆ postBeginJob()

void evf::FastMonitoringService::postBeginJob ( )

Definition at line 466 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

466  {
467  std::string&& moduleLegStrJson = makeModuleLegendaJson();
468  FileIO::writeStringToFile(moduleLegendFileJson_, moduleLegStrJson);
469 
470  std::string inputLegendStrJson = makeInputLegendaJson();
471  FileIO::writeStringToFile(inputLegendFileJson_, inputLegendStrJson);
472 
473  std::string pathLegendStrJson = makePathLegendaJson();
474  FileIO::writeStringToFile(pathLegendFileJson_, pathLegendStrJson);
475 
476  fmt_->m_data.macrostate_ = FastMonState::sJobReady;
477 
478  //update number of entries in module histogram
479  std::lock_guard<std::mutex> lock(fmt_->monlock_);
480  //double the size to add post-acquire states
481  fmt_->m_data.microstateBins_ = fmt_->m_data.encModule_.vecsize() * 2;
482  }
std::shared_ptr< FastMonitoringThread > fmt_

◆ postEndJob()

void evf::FastMonitoringService::postEndJob ( )

Definition at line 484 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

484  {
485  fmt_->m_data.macrostate_ = FastMonState::sJobEnded;
486  fmt_->stop();
487  }
std::shared_ptr< FastMonitoringThread > fmt_

◆ postEvent()

void evf::FastMonitoringService::postEvent ( edm::StreamContext const &  sc)

Definition at line 637 of file FastMonitoringService.cc.

References fmt_, evf::FastMonState::mIdle, nopath_, reservedMicroStateNames, edm::StreamContext::streamID(), and totalEventsProcessed_.

Referenced by FastMonitoringService().

637  {
638  fmt_->m_data.microstate_[sc.streamID()] = &reservedMicroStateNames[FastMonState::mIdle];
639 
640  fmt_->m_data.ministate_[sc.streamID()] = &nopath_;
641 
642  (*(fmt_->m_data.processed_[sc.streamID()]))++;
643 
644  //fast path counter (events accumulated in a run)
645  unsigned long res = totalEventsProcessed_.fetch_add(1, std::memory_order_relaxed);
646  fmt_->m_data.fastPathProcessedJ_ = res + 1;
647  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
Definition: Electron.h:6
static const std::string nopath_
std::atomic< unsigned long > totalEventsProcessed_
std::shared_ptr< FastMonitoringThread > fmt_

◆ postGlobalBeginRun()

void evf::FastMonitoringService::postGlobalBeginRun ( edm::GlobalContext const &  gc)

Definition at line 489 of file FastMonitoringService.cc.

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

489  {
490  fmt_->m_data.macrostate_ = FastMonState::sRunning;
491  isInitTransition_ = false;
492  }
std::atomic< bool > isInitTransition_
std::shared_ptr< FastMonitoringThread > fmt_

◆ postGlobalEndLumi()

void evf::FastMonitoringService::postGlobalEndLumi ( edm::GlobalContext const &  gc)

Definition at line 587 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

587  {
588  std::lock_guard<std::mutex> lock(fmt_->monlock_);
589  unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
590  //LS monitoring snapshot with input source data has been taken in previous callback
591  avgLeadTime_.erase(lumi);
593  lockStatsDuringLumi_.erase(lumi);
594 
595  //output module already used this in end lumi (this could be migrated to EvFDaqDirector as it is essential for FFF bookkeeping)
597  }
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
std::map< unsigned int, double > avgLeadTime_
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
std::shared_ptr< FastMonitoringThread > fmt_

◆ postModuleEvent()

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

Definition at line 673 of file FastMonitoringService.cc.

References fmt_, evf::FastMonState::mFwkOvhMod, reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

673  {
674  fmt_->m_data.microstate_[sc.streamID().value()] = &reservedMicroStateNames[FastMonState::mFwkOvhMod];
675  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
std::shared_ptr< FastMonitoringThread > fmt_

◆ postModuleEventAcquire()

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

Definition at line 662 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

663  {
664  //fmt_->m_data.microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
665  fmt_->m_data.microstateAcqFlag_[sc.streamID().value()] = 1;
666  }
std::shared_ptr< FastMonitoringThread > fmt_

◆ postSourceEvent()

void evf::FastMonitoringService::postSourceEvent ( edm::StreamID  sid)

Definition at line 653 of file FastMonitoringService.cc.

References fmt_, evf::FastMonState::mFwkOvhSrc, reservedMicroStateNames, and edm::StreamID::value().

Referenced by FastMonitoringService().

653  {
654  fmt_->m_data.microstate_[sid.value()] = &reservedMicroStateNames[FastMonState::mFwkOvhSrc];
655  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
unsigned int value() const
Definition: StreamID.h:43
std::shared_ptr< FastMonitoringThread > fmt_

◆ postStreamBeginLumi()

void evf::FastMonitoringService::postStreamBeginLumi ( edm::StreamContext const &  sc)

Definition at line 612 of file FastMonitoringService.cc.

References fmt_, evf::FastMonState::mIdle, reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

612  {
613  fmt_->m_data.microstate_[sc.streamID().value()] = &reservedMicroStateNames[FastMonState::mIdle];
614  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
std::shared_ptr< FastMonitoringThread > fmt_

◆ postStreamEndLumi()

void evf::FastMonitoringService::postStreamEndLumi ( edm::StreamContext const &  sc)

Definition at line 627 of file FastMonitoringService.cc.

References fmt_, evf::FastMonState::mFwkEoL, reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

627  {
628  fmt_->m_data.microstate_[sc.streamID().value()] = &reservedMicroStateNames[FastMonState::mFwkEoL];
629  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
std::shared_ptr< FastMonitoringThread > fmt_

◆ preallocate()

void evf::FastMonitoringService::preallocate ( edm::service::SystemBounds const &  bounds)

Definition at line 243 of file FastMonitoringService.cc.

References nStreams_, and nThreads_.

Referenced by FastMonitoringService().

243  {
244  nStreams_ = bounds.maxNumberOfStreams();
245  nThreads_ = bounds.maxNumberOfThreads();
246  //this should already be >=1
247  if (nStreams_ == 0)
248  nStreams_ = 1;
249  if (nThreads_ == 0)
250  nThreads_ = 1;
251  }

◆ preBeginJob()

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

Definition at line 253 of file FastMonitoringService.cc.

References edm::PathsAndConsumesOfModulesBase::endPaths(), Exception, fastMicrostateDefPath_, fastName_, fastPath_, fastPathList_, filePerFwkStream_, fmt_, mps_fire::i, evf::FastMonState::inCOUNT, inputLegendFileJson_, isInitTransition_, lastGlobalLumi_, LogDebug, lumiFromSource_, evf::FastMonState::mCOUNT, evf::FastMonState::MCOUNT, microstateDefPath_, evf::FastMonState::mInvalid, moduleLegendFile_, moduleLegendFileJson_, monInit_, nopath_, nStreams_, nThreads_, Utilities::operator, castor_dqm_sourceclient_file_cfg::path, pathLegendFile_, pathLegendFileJson_, edm::PathsAndConsumesOfModulesBase::paths(), reservedMicroStateNames, runDirectory_, evf::FastMonState::sInit, sleepTime_, snapshotRunner(), streamCounterUpdating_, threadIDAvailable_, and workingDirectory_.

Referenced by FastMonitoringService().

254  {
255  // FIND RUN DIRECTORY
256  // The run dir should be set via the configuration of EvFDaqDirector
257 
258  if (edm::Service<evf::EvFDaqDirector>().operator->() == nullptr) {
259  throw cms::Exception("FastMonitoringService") << "EvFDaqDirector is not present";
260  }
261  std::filesystem::path runDirectory{edm::Service<evf::EvFDaqDirector>()->baseRunDir()};
262  workingDirectory_ = runDirectory_ = runDirectory;
263  workingDirectory_ /= "mon";
264 
265  if (!std::filesystem::is_directory(workingDirectory_)) {
266  LogDebug("FastMonitoringService") << "<MON> DIR NOT FOUND! Trying to create -: " << workingDirectory_.string();
267  std::filesystem::create_directories(workingDirectory_);
268  if (!std::filesystem::is_directory(workingDirectory_))
269  edm::LogWarning("FastMonitoringService") << "Unable to create <MON> DIR -: " << workingDirectory_.string()
270  << ". No monitoring data will be written.";
271  }
272 
273  std::ostringstream fastFileName;
274 
275  fastFileName << fastName_ << "_pid" << std::setfill('0') << std::setw(5) << getpid() << ".fast";
277  fast /= fastFileName.str();
278  fastPath_ = fast.string();
279  if (filePerFwkStream_)
280  for (unsigned int i = 0; i < nStreams_; i++) {
281  std::ostringstream fastFileNameTid;
282  fastFileNameTid << fastName_ << "_pid" << std::setfill('0') << std::setw(5) << getpid() << "_tid" << i
283  << ".fast";
285  fastTid /= fastFileNameTid.str();
286  fastPathList_.push_back(fastTid.string());
287  }
288 
289  std::ostringstream moduleLegFile;
290  std::ostringstream moduleLegFileJson;
291  moduleLegFile << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
292  moduleLegFileJson << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
293  moduleLegendFile_ = (workingDirectory_ / moduleLegFile.str()).string();
294  moduleLegendFileJson_ = (workingDirectory_ / moduleLegFileJson.str()).string();
295 
296  std::ostringstream pathLegFile;
297  std::ostringstream pathLegFileJson;
298  pathLegFile << "pathlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
299  pathLegendFile_ = (workingDirectory_ / pathLegFile.str()).string();
300  pathLegFileJson << "pathlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
301  pathLegendFileJson_ = (workingDirectory_ / pathLegFileJson.str()).string();
302 
303  std::ostringstream inputLegFileJson;
304  inputLegFileJson << "inputlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
305  inputLegendFileJson_ = (workingDirectory_ / inputLegFileJson.str()).string();
306 
307  LogDebug("FastMonitoringService") << "Initializing FastMonitor with microstate def path -: " << microstateDefPath_;
308  //<< encPath_.current_ + 1 << " " << encModule_.current_ + 1
309 
310  /*
311  * initialize the fast monitor with:
312  * vector of pointers to monitorable parameters
313  * path to definition
314  *
315  */
316 
317  fmt_->m_data.macrostate_ = FastMonState::sInit;
318 
319  for (unsigned int i = 0; i < (FastMonState::mCOUNT); i++)
320  fmt_->m_data.encModule_.updateReserved(static_cast<const void*>(reservedMicroStateNames + i));
321  fmt_->m_data.encModule_.completeReservedWithDummies();
322 
323  for (unsigned int i = 0; i < nStreams_; i++) {
324  fmt_->m_data.ministate_.emplace_back(&nopath_);
325  fmt_->m_data.microstate_.emplace_back(&reservedMicroStateNames[FastMonState::mInvalid]);
326  fmt_->m_data.microstateAcqFlag_.push_back(0);
327 
328  //for synchronization
329  streamCounterUpdating_.push_back(new std::atomic<bool>(false));
330 
331  //path (mini) state
332  fmt_->m_data.encPath_.emplace_back(0);
333  fmt_->m_data.encPath_[i].update(static_cast<const void*>(&nopath_));
334 
335  for (auto& path : pathsInfo.paths()) {
336  fmt_->m_data.encPath_[i].updatePreinit(path);
337  }
338  for (auto& endPath : pathsInfo.endPaths()) {
339  fmt_->m_data.encPath_[i].updatePreinit(endPath);
340  }
341  }
342  //for (unsigned int i=0;i<nThreads_;i++)
343  // threadMicrostate_.push_back(&reservedMicroStateNames[mInvalid]);
344 
345  //initial size until we detect number of bins
346  fmt_->m_data.macrostateBins_ = FastMonState::MCOUNT;
347  fmt_->m_data.microstateBins_ = 0;
348  fmt_->m_data.inputstateBins_ = FastMonState::inCOUNT;
349  fmt_->m_data.ministateBins_ = fmt_->m_data.encPath_[0].vecsize();
350 
351  lastGlobalLumi_ = 0;
352  isInitTransition_ = true;
353  lumiFromSource_ = 0;
354 
355  //startup monitoring
356  fmt_->resetFastMonitor(microstateDefPath_, fastMicrostateDefPath_);
357  fmt_->jsonMonitor_->setNStreams(nStreams_);
358  fmt_->m_data.registerVariables(fmt_->jsonMonitor_.get(), nStreams_, threadIDAvailable_ ? nThreads_ : 0);
359  monInit_.store(false, std::memory_order_release);
360  if (sleepTime_ > 0)
362 
363  //this definition needs: #include "tbb/compat/thread"
364  //however this would results in TBB imeplementation replacing std::thread
365  //(both supposedly call pthread_self())
366  //number of threads created in process could be obtained from /proc,
367  //assuming that all posix threads are true kernel threads capable of running in parallel
368 
369  //#if TBB_IMPLEMENT_CPP0X
371  //threadIDAvailable_=true;
372  //#endif
373  }
std::atomic< bool > isInitTransition_
std::filesystem::path workingDirectory_
std::vector< std::atomic< bool > * > streamCounterUpdating_
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
static const std::string nopath_
std::filesystem::path runDirectory_
std::vector< std::string > fastPathList_
std::shared_ptr< FastMonitoringThread > fmt_
Log< level::Warning, false > LogWarning
#define LogDebug(id)

◆ preEvent()

void evf::FastMonitoringService::preEvent ( edm::StreamContext const &  sc)

Definition at line 635 of file FastMonitoringService.cc.

Referenced by FastMonitoringService().

635 {}

◆ preGlobalBeginLumi()

void evf::FastMonitoringService::preGlobalBeginLumi ( edm::GlobalContext const &  gc)

Definition at line 494 of file FastMonitoringService.cc.

References fmt_, lastGlobalLumi_, CommonMethods::lock(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), and lumiStartTime_.

Referenced by FastMonitoringService().

494  {
495  timeval lumiStartTime;
496  gettimeofday(&lumiStartTime, nullptr);
497  unsigned int newLumi = gc.luminosityBlockID().luminosityBlock();
498  lastGlobalLumi_ = newLumi;
499 
500  std::lock_guard<std::mutex> lock(fmt_->monlock_);
501  lumiStartTime_[newLumi] = lumiStartTime;
502  }
std::map< unsigned int, timeval > lumiStartTime_
std::shared_ptr< FastMonitoringThread > fmt_

◆ preGlobalEarlyTermination()

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

Definition at line 391 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().

391  {
394  context = " FromThisContext ";
396  context = " FromAnotherContext";
398  context = " FromExternalSignal";
399  edm::LogWarning("FastMonitoringService")
400  << " GLOBAL "
401  << "earlyTermination -: LS:" << gc.luminosityBlockID().luminosityBlock() << " " << context;
402  std::lock_guard<std::mutex> lock(fmt_->monlock_);
403  exceptionInLS_.push_back(gc.luminosityBlockID().luminosityBlock());
404  has_data_exception_.store(true);
405  }
std::atomic< bool > has_data_exception_
std::shared_ptr< FastMonitoringThread > fmt_
Log< level::Warning, false > LogWarning
std::vector< unsigned int > exceptionInLS_

◆ preGlobalEndLumi()

void evf::FastMonitoringService::preGlobalEndLumi ( edm::GlobalContext const &  gc)

Definition at line 504 of file FastMonitoringService.cc.

References accuSize_, doSnapshot(), Exception, exception_detected_, exceptionInLS_, filePerFwkStream_, fmt_, FedRawDataInputSource::getEventReport(), inputSource_, CommonMethods::lock(), LogDebug, BXlumiParameters_cfi::lumi, edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), lumiStartTime_, convertSQLitetoXML_cfg::output, castor_dqm_sourceclient_file_cfg::path, processedEventsPerLumi_, edm::shutdown_flag, sleepTime_, slowName_, throughputFactor(), jsoncollector::IntJ::value(), and workingDirectory_.

Referenced by FastMonitoringService().

504  {
505  unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
506  LogDebug("FastMonitoringService") << "Lumi ended. Writing JSON information. LUMI -: " << lumi;
507  timeval lumiStopTime;
508  gettimeofday(&lumiStopTime, nullptr);
509 
510  std::lock_guard<std::mutex> lock(fmt_->monlock_);
511 
512  // Compute throughput
513  timeval stt = lumiStartTime_[lumi];
514  lumiStartTime_.erase(lumi);
515  unsigned long usecondsForLumi = (lumiStopTime.tv_sec - stt.tv_sec) * 1000000 + (lumiStopTime.tv_usec - stt.tv_usec);
516  unsigned long accuSize = accuSize_.find(lumi) == accuSize_.end() ? 0 : accuSize_[lumi];
517  accuSize_.erase(lumi);
518  double throughput = throughputFactor() * double(accuSize) / double(usecondsForLumi);
519  //store to registered variable
520  fmt_->m_data.fastThroughputJ_.value() = throughput;
521 
522  //update
523  doSnapshot(lumi, true);
524 
525  //retrieve one result we need (todo: sanity check if it's found)
526  IntJ* lumiProcessedJptr = dynamic_cast<IntJ*>(fmt_->jsonMonitor_->getMergedIntJForLumi("Processed", lumi));
527  if (!lumiProcessedJptr)
528  throw cms::Exception("FastMonitoringService") << "Internal error: got null pointer from FastMonitor";
529  processedEventsPerLumi_[lumi] = std::pair<unsigned int, bool>(lumiProcessedJptr->value(), false);
530 
531  //checking if exception has been thrown (in case of Global/Stream early termination, for this LS)
532  bool exception_detected = exception_detected_;
533  for (auto ex : exceptionInLS_)
534  if (lumi == ex)
535  exception_detected = true;
536 
537  if (edm::shutdown_flag || exception_detected) {
538  edm::LogInfo("FastMonitoringService")
539  << "Run interrupted. Skip writing EoL information -: " << processedEventsPerLumi_[lumi].first
540  << " events were processed in LUMI " << lumi;
541  //this will prevent output modules from producing json file for possibly incomplete lumi
542  processedEventsPerLumi_[lumi].first = 0;
543  processedEventsPerLumi_[lumi].second = true;
544  //disable this exception, so service can be used standalone (will be thrown if output module asks for this information)
545  //throw cms::Exception("FastMonitoringService") << "SOURCE did not send update for lumi block. LUMI -:" << lumi;
546  return;
547  }
548 
549  if (inputSource_) {
550  auto sourceReport = inputSource_->getEventReport(lumi, true);
551  if (sourceReport.first) {
552  if (sourceReport.second != processedEventsPerLumi_[lumi].first) {
553  throw cms::Exception("FastMonitoringService") << "MISMATCH with SOURCE update. LUMI -: " << lumi
554  << ", events(processed):" << processedEventsPerLumi_[lumi].first
555  << " events(source):" << sourceReport.second;
556  }
557  }
558  }
559  edm::LogInfo("FastMonitoringService")
560  << "Statistics for lumisection -: lumi = " << lumi << " events = " << lumiProcessedJptr->value()
561  << " time = " << usecondsForLumi / 1000000 << " size = " << accuSize << " thr = " << throughput;
562  delete lumiProcessedJptr;
563 
564  //full global and stream merge&output for this lumi
565 
566  // create file name for slow monitoring file
567  bool output = sleepTime_ > 0;
568  if (filePerFwkStream_) {
569  std::stringstream slowFileNameStem;
570  slowFileNameStem << slowName_ << "_ls" << std::setfill('0') << std::setw(4) << lumi << "_pid" << std::setfill('0')
571  << std::setw(5) << getpid();
573  slow /= slowFileNameStem.str();
574  fmt_->jsonMonitor_->outputFullJSONs(slow.string(), ".jsn", lumi, output);
575  } else {
576  std::stringstream slowFileName;
577  slowFileName << slowName_ << "_ls" << std::setfill('0') << std::setw(4) << lumi << "_pid" << std::setfill('0')
578  << std::setw(5) << getpid() << ".jsn";
580  slow /= slowFileName.str();
581  //full global and stream merge and JSON write for this lumi
582  fmt_->jsonMonitor_->outputFullJSON(slow.string(), lumi, output);
583  }
584  fmt_->jsonMonitor_->discardCollected(lumi); //we don't do further updates for this lumi
585  }
std::map< unsigned int, timeval > lumiStartTime_
std::pair< bool, unsigned int > getEventReport(unsigned int lumi, bool erase)
std::map< unsigned int, unsigned long > accuSize_
std::filesystem::path workingDirectory_
volatile std::atomic< bool > shutdown_flag
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
constexpr double throughputFactor()
Log< level::Info, false > LogInfo
FedRawDataInputSource * inputSource_
Definition: output.py:1
std::shared_ptr< FastMonitoringThread > fmt_
std::vector< unsigned int > exceptionInLS_
#define LogDebug(id)

◆ preModuleBeginJob()

void evf::FastMonitoringService::preModuleBeginJob ( edm::ModuleDescription const &  desc)

Definition at line 451 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

451  {
452  std::lock_guard<std::mutex> lock(fmt_->monlock_);
453  //std::cout << " Pre module Begin Job module: " << desc.moduleName() << std::endl;
454 
455  //build a map of modules keyed by their module description address
456  //here we need to treat output modules in a special way so they can be easily singled out
457  if (desc.moduleName() == "Stream" || desc.moduleName() == "GlobalEvFOutputModule" ||
458  desc.moduleName() == "EvFOutputModule" || desc.moduleName() == "EventStreamFileWriter" ||
459  desc.moduleName() == "PoolOutputModule") {
460  fmt_->m_data.encModule_.updateReserved((void*)&desc);
461  nOutputModules_++;
462  } else
463  fmt_->m_data.encModule_.update((void*)&desc);
464  }
std::shared_ptr< FastMonitoringThread > fmt_

◆ preModuleEvent()

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

Definition at line 668 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

668  {
669  fmt_->m_data.microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
670  fmt_->m_data.microstateAcqFlag_[sc.streamID().value()] = 0;
671  }
std::shared_ptr< FastMonitoringThread > fmt_

◆ preModuleEventAcquire()

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

Definition at line 657 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

658  {
659  fmt_->m_data.microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
660  }
std::shared_ptr< FastMonitoringThread > fmt_

◆ prePathEvent()

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

Definition at line 631 of file FastMonitoringService.cc.

References fmt_, edm::PathContext::pathName(), and edm::StreamContext::streamID().

Referenced by FastMonitoringService().

631  {
632  fmt_->m_data.ministate_[sc.streamID()] = &(pc.pathName());
633  }
std::shared_ptr< FastMonitoringThread > fmt_

◆ preSourceEarlyTermination()

void evf::FastMonitoringService::preSourceEarlyTermination ( edm::TerminationOrigin  to)

Definition at line 407 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().

407  {
410  context = " FromThisContext ";
412  context = " FromAnotherContext";
414  context = " FromExternalSignal";
415  edm::LogWarning("FastMonitoringService") << " SOURCE "
416  << "earlyTermination -: " << context;
417  std::lock_guard<std::mutex> lock(fmt_->monlock_);
418  exception_detected_ = true;
419  has_source_exception_.store(true);
420  has_data_exception_.store(true);
421  }
std::atomic< bool > has_data_exception_
std::atomic< bool > has_source_exception_
std::shared_ptr< FastMonitoringThread > fmt_
Log< level::Warning, false > LogWarning

◆ preSourceEvent()

void evf::FastMonitoringService::preSourceEvent ( edm::StreamID  sid)

Definition at line 649 of file FastMonitoringService.cc.

References fmt_, evf::FastMonState::mInput, reservedMicroStateNames, and edm::StreamID::value().

Referenced by FastMonitoringService().

649  {
650  fmt_->m_data.microstate_[sid.value()] = &reservedMicroStateNames[FastMonState::mInput];
651  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
unsigned int value() const
Definition: StreamID.h:43
std::shared_ptr< FastMonitoringThread > fmt_

◆ preStreamBeginLumi()

void evf::FastMonitoringService::preStreamBeginLumi ( edm::StreamContext const &  sc)

Definition at line 599 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), fmt_, CommonMethods::lock(), edm::EventID::luminosityBlock(), evf::FastMonState::mBoL, nopath_, reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

599  {
600  unsigned int sid = sc.streamID().value();
601 
602  std::lock_guard<std::mutex> lock(fmt_->monlock_);
603  fmt_->m_data.streamLumi_[sid] = sc.eventID().luminosityBlock();
604 
605  //reset collected values for this stream
606  *(fmt_->m_data.processed_[sid]) = 0;
607 
608  fmt_->m_data.ministate_[sid] = &nopath_;
609  fmt_->m_data.microstate_[sid] = &reservedMicroStateNames[FastMonState::mBoL];
610  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
static const std::string nopath_
std::shared_ptr< FastMonitoringThread > fmt_

◆ preStreamEarlyTermination()

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

Definition at line 375 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().

375  {
378  context = " FromThisContext ";
380  context = " FromAnotherContext";
382  context = " FromExternalSignal";
383  edm::LogWarning("FastMonitoringService")
384  << " STREAM " << sc.streamID().value() << " earlyTermination -: ID:" << sc.eventID()
385  << " LS:" << sc.eventID().luminosityBlock() << " " << context;
386  std::lock_guard<std::mutex> lock(fmt_->monlock_);
387  exceptionInLS_.push_back(sc.eventID().luminosityBlock());
388  has_data_exception_.store(true);
389  }
std::atomic< bool > has_data_exception_
std::shared_ptr< FastMonitoringThread > fmt_
Log< level::Warning, false > LogWarning
std::vector< unsigned int > exceptionInLS_

◆ preStreamEndLumi()

void evf::FastMonitoringService::preStreamEndLumi ( edm::StreamContext const &  sc)

Definition at line 616 of file FastMonitoringService.cc.

References edm::StreamContext::eventID(), fmt_, CommonMethods::lock(), edm::EventID::luminosityBlock(), evf::FastMonState::mEoL, nopath_, reservedMicroStateNames, edm::StreamContext::streamID(), and edm::StreamID::value().

Referenced by FastMonitoringService().

616  {
617  unsigned int sid = sc.streamID().value();
618  std::lock_guard<std::mutex> lock(fmt_->monlock_);
619 
620  //update processed count to be complete at this time
621  //doStreamEOLSnapshot(sc.eventID().luminosityBlock(), sid);
622  fmt_->jsonMonitor_->snapStreamAtomic(sc.eventID().luminosityBlock(), sid);
623  //reset this in case stream does not get notified of next lumi (we keep processed events only)
624  fmt_->m_data.ministate_[sid] = &nopath_;
625  fmt_->m_data.microstate_[sid] = &reservedMicroStateNames[FastMonState::mEoL];
626  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
static const std::string nopath_
std::shared_ptr< FastMonitoringThread > fmt_

◆ reportLockWait()

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

Definition at line 742 of file FastMonitoringService.cc.

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

Referenced by FedRawDataInputSource::readSupervisor().

742  {
743  std::lock_guard<std::mutex> lock(fmt_->monlock_);
744  lockStatsDuringLumi_[ls] = std::pair<double, unsigned int>(waitTime, lockCount);
745  }
def ls(path, rec=False)
Definition: eostools.py:349
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
std::shared_ptr< FastMonitoringThread > fmt_

◆ setExceptionDetected()

void evf::FastMonitoringService::setExceptionDetected ( unsigned int  ls)

Definition at line 423 of file FastMonitoringService.cc.

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

Referenced by FedRawDataInputSource::getNextEvent().

423  {
424  std::lock_guard<std::mutex> lock(fmt_->monlock_);
425  if (!ls)
426  exception_detected_ = true;
427  else
428  exceptionInLS_.push_back(ls);
429  }
def ls(path, rec=False)
Definition: eostools.py:349
std::shared_ptr< FastMonitoringThread > fmt_
std::vector< unsigned int > exceptionInLS_

◆ setInputSource()

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

Definition at line 221 of file FastMonitoringService.h.

References inputSource_.

Referenced by FedRawDataInputSource::FedRawDataInputSource().

221 { inputSource_ = inputSource; }
FedRawDataInputSource * inputSource_

◆ setInState()

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

Definition at line 222 of file FastMonitoringService.h.

References inputState_.

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

222 { inputState_ = inputState; }
std::atomic< FastMonState::InputState > inputState_

◆ setInStateSup()

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

Definition at line 223 of file FastMonitoringService.h.

References inputSupervisorState_.

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

223 { inputSupervisorState_ = inputState; }
std::atomic< FastMonState::InputState > inputSupervisorState_

◆ setMicroState() [1/2]

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

Definition at line 681 of file FastMonitoringService.cc.

References fmt_, mps_fire::i, visualization-live-secondInstance_cfg::m, nStreams_, and reservedMicroStateNames.

681  {
682  for (unsigned int i = 0; i < nStreams_; i++)
683  fmt_->m_data.microstate_[i] = &reservedMicroStateNames[m];
684  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
std::shared_ptr< FastMonitoringThread > fmt_

◆ setMicroState() [2/2]

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

Definition at line 687 of file FastMonitoringService.cc.

References fmt_, visualization-live-secondInstance_cfg::m, and reservedMicroStateNames.

687  {
688  fmt_->m_data.microstate_[sid] = &reservedMicroStateNames[m];
689  }
static const edm::ModuleDescription reservedMicroStateNames[FastMonState::mCOUNT]
std::shared_ptr< FastMonitoringThread > fmt_

◆ shouldWriteFiles()

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

Definition at line 214 of file FastMonitoringService.h.

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

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

214  {
215  unsigned int processed = getEventsProcessedForLumi(lumi);
216  if (proc)
217  *proc = processed;
218  return !getAbortFlagForLumi(lumi);
219  }
unsigned int getEventsProcessedForLumi(unsigned int lumi, bool *abortFlag=nullptr)
bool getAbortFlagForLumi(unsigned int lumi)

◆ snapshotRunner()

void evf::FastMonitoringService::snapshotRunner ( )
private

Definition at line 782 of file FastMonitoringService.cc.

References doSnapshot(), relativeConstraints::empty, f, fastMonIntervals_, fastPath_, fastPathList_, filePerFwkStream_, fmt_, mps_fire::i, inputState_, inputStateNames, inputSupervisorState_, lastGlobalLumi_, CommonMethods::lock(), monInit_, mps_check::msg, nStreams_, AlCaHLTBitMon_ParallelJobs::p, sleepTime_, snapCounter_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by preBeginJob().

782  {
783  monInit_.exchange(true, std::memory_order_acquire);
784  while (!fmt_->m_stoprequest) {
785  std::vector<std::vector<unsigned int>> lastEnc;
786  {
787  std::unique_lock<std::mutex> lock(fmt_->monlock_);
788 
789  doSnapshot(lastGlobalLumi_, false);
790 
791  lastEnc.emplace_back(fmt_->m_data.ministateEncoded_);
792  lastEnc.emplace_back(fmt_->m_data.microstateEncoded_);
793 
795  if (filePerFwkStream_) {
796  std::vector<std::string> CSVv;
797  for (unsigned int i = 0; i < nStreams_; i++) {
798  CSVv.push_back(fmt_->jsonMonitor_->getCSVString((int)i));
799  }
800  // release mutex before writing out fast path file
801  lock.release()->unlock();
802  for (unsigned int i = 0; i < nStreams_; i++) {
803  if (!CSVv[i].empty())
804  fmt_->jsonMonitor_->outputCSV(fastPathList_[i], CSVv[i]);
805  }
806  } else {
807  std::string CSV = fmt_->jsonMonitor_->getCSVString();
808  // release mutex before writing out fast path file
809  lock.release()->unlock();
810  if (!CSV.empty())
811  fmt_->jsonMonitor_->outputCSV(fastPath_, CSV);
812  }
813  }
814  snapCounter_++;
815  }
816 
817  {
818  edm::LogInfo msg("FastMonitoringService");
819  auto f = [&](std::vector<unsigned int> const& p) {
820  for (unsigned int i = 0; i < nStreams_; i++) {
821  if (i == 0)
822  msg << "[" << p[i] << ",";
823  else if (i <= nStreams_ - 1)
824  msg << p[i] << ",";
825  else
826  msg << p[i] << "]";
827  }
828  };
829 
830  msg << "Current states: Ms=" << fmt_->m_data.fastMacrostateJ_.value() << " ms=";
831  f(lastEnc[0]);
832  msg << " us=";
833  f(lastEnc[1]);
835  }
836 
837  ::sleep(sleepTime_);
838  }
839  }
std::atomic< FastMonState::InputState > inputState_
static const std::string inputStateNames[FastMonState::inCOUNT]
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
std::atomic< FastMonState::InputState > inputSupervisorState_
double f[11][100]
std::vector< std::string > fastPathList_
tuple msg
Definition: mps_check.py:286
std::shared_ptr< FastMonitoringThread > fmt_

◆ startedLookingForFile()

void evf::FastMonitoringService::startedLookingForFile ( )

Definition at line 706 of file FastMonitoringService.cc.

References fileLookStart_.

Referenced by FedRawDataInputSource::readSupervisor().

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

◆ stoppedLookingForFile()

void evf::FastMonitoringService::stoppedLookingForFile ( unsigned int  lumi)

Definition at line 714 of file FastMonitoringService.cc.

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

Referenced by FedRawDataInputSource::readSupervisor().

714  {
715  gettimeofday(&fileLookStop_, nullptr);
716  /*
717  std::cout << "Stopped looking for .raw file at: s=" << fileLookStop_.tv_sec << ": ms = "
718  << fileLookStop_.tv_usec / 1000.0 << std::endl;
719  */
720  std::lock_guard<std::mutex> lock(fmt_->monlock_);
721 
722  if (lumi > lumiFromSource_) {
724  leadTimes_.clear();
725  }
726  unsigned long elapsedTime = (fileLookStop_.tv_sec - fileLookStart_.tv_sec) * 1000000 // sec to us
727  + (fileLookStop_.tv_usec - fileLookStart_.tv_usec); // us
728  // add this to lead times for this lumi
729  leadTimes_.push_back((double)elapsedTime);
730 
731  // recompute average lead time for this lumi
732  if (leadTimes_.size() == 1)
734  else {
735  double totTime = 0;
736  for (unsigned int i = 0; i < leadTimes_.size(); i++)
737  totTime += leadTimes_[i];
738  avgLeadTime_[lumi] = 0.001 * (totTime / leadTimes_.size());
739  }
740  }
std::map< unsigned int, double > avgLeadTime_
std::vector< double > leadTimes_
std::shared_ptr< FastMonitoringThread > fmt_

Member Data Documentation

◆ accuSize_

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

Definition at line 261 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), and preGlobalEndLumi().

◆ avgLeadTime_

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

Definition at line 258 of file FastMonitoringService.h.

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

◆ collectedPathList_

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

Definition at line 272 of file FastMonitoringService.h.

◆ exception_detected_

bool evf::FastMonitoringService::exception_detected_ = false
private

◆ exceptionInLS_

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

◆ fastMicrostateDefPath_

std::string evf::FastMonitoringService::fastMicrostateDefPath_
private

Definition at line 243 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

◆ fastMonIntervals_

unsigned int evf::FastMonitoringService::fastMonIntervals_
private

Definition at line 241 of file FastMonitoringService.h.

Referenced by snapshotRunner().

◆ fastName_

std::string evf::FastMonitoringService::fastName_
private

Definition at line 244 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ fastPath_

std::string evf::FastMonitoringService::fastPath_
private

Definition at line 244 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ fastPathList_

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

Definition at line 293 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ fileLookStart_

timeval evf::FastMonitoringService::fileLookStart_
private

Definition at line 250 of file FastMonitoringService.h.

Referenced by startedLookingForFile(), and stoppedLookingForFile().

◆ fileLookStop_

timeval evf::FastMonitoringService::fileLookStop_
private

Definition at line 250 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

◆ filePerFwkStream_

bool evf::FastMonitoringService::filePerFwkStream_
private

Definition at line 245 of file FastMonitoringService.h.

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

◆ filesProcessedDuringLumi_

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

Definition at line 259 of file FastMonitoringService.h.

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

◆ fmt_

std::shared_ptr<FastMonitoringThread> evf::FastMonitoringService::fmt_
private

◆ has_data_exception_

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

◆ has_source_exception_

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

◆ inputLegendFileJson_

std::string evf::FastMonitoringService::inputLegendFileJson_
private

Definition at line 285 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

◆ inputSource_

FedRawDataInputSource* evf::FastMonitoringService::inputSource_ = nullptr
private

Definition at line 234 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

◆ inputState_

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

Definition at line 235 of file FastMonitoringService.h.

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

◆ inputStateNames

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

Definition at line 160 of file FastMonitoringService.h.

Referenced by makeInputLegendaJson(), and snapshotRunner().

◆ inputSupervisorState_

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

Definition at line 236 of file FastMonitoringService.h.

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

◆ isInitTransition_

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

Definition at line 253 of file FastMonitoringService.h.

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

◆ lastGlobalLumi_

unsigned int evf::FastMonitoringService::lastGlobalLumi_
private

Definition at line 252 of file FastMonitoringService.h.

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

◆ leadTimes_

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

Definition at line 262 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

◆ lockStatsDuringLumi_

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

Definition at line 263 of file FastMonitoringService.h.

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

◆ lumiFromSource_

unsigned int evf::FastMonitoringService::lumiFromSource_
private

Definition at line 254 of file FastMonitoringService.h.

Referenced by preBeginJob(), and stoppedLookingForFile().

◆ lumiStartTime_

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

Definition at line 249 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 159 of file FastMonitoringService.h.

◆ microstateDefPath_

std::string evf::FastMonitoringService::microstateDefPath_
private

Definition at line 243 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

◆ moduleLegendFile_

std::string evf::FastMonitoringService::moduleLegendFile_
private

Definition at line 281 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ moduleLegendFileJson_

std::string evf::FastMonitoringService::moduleLegendFileJson_
private

Definition at line 282 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

◆ monInit_

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

Definition at line 288 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ nopath_

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

◆ nOutputModules_

unsigned int evf::FastMonitoringService::nOutputModules_ = 0
private

Definition at line 286 of file FastMonitoringService.h.

Referenced by makeModuleLegendaJson(), and preModuleBeginJob().

◆ nStreams_

unsigned int evf::FastMonitoringService::nStreams_
private

◆ nThreads_

unsigned int evf::FastMonitoringService::nThreads_
private

Definition at line 239 of file FastMonitoringService.h.

Referenced by preallocate(), and preBeginJob().

◆ pathLegendFile_

std::string evf::FastMonitoringService::pathLegendFile_
private

Definition at line 283 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ pathLegendFileJson_

std::string evf::FastMonitoringService::pathLegendFileJson_
private

Definition at line 284 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

◆ pathNamesReady_

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

Definition at line 273 of file FastMonitoringService.h.

◆ processedEventsPerLumi_

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

◆ reservedMicroStateNames

const edm::ModuleDescription evf::FastMonitoringService::reservedMicroStateNames
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")}

Definition at line 158 of file FastMonitoringService.h.

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

◆ runDirectory_

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

Definition at line 275 of file FastMonitoringService.h.

Referenced by getRunDirName(), and preBeginJob().

◆ sleepTime_

int evf::FastMonitoringService::sleepTime_
private

Definition at line 240 of file FastMonitoringService.h.

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

◆ slowName_

std::string evf::FastMonitoringService::slowName_
private

Definition at line 244 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi().

◆ snapCounter_

unsigned int evf::FastMonitoringService::snapCounter_ = 0
private

Definition at line 242 of file FastMonitoringService.h.

Referenced by snapshotRunner().

◆ streamCounterUpdating_

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

Definition at line 270 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ threadIDAvailable_

bool evf::FastMonitoringService::threadIDAvailable_ = false
private

Definition at line 277 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ totalEventsProcessed_

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

Definition at line 279 of file FastMonitoringService.h.

Referenced by postEvent().

◆ workingDirectory_

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

Definition at line 275 of file FastMonitoringService.h.

Referenced by preBeginJob(), and preGlobalEndLumi().