CMS 3D CMS Logo

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

#include <FastMonitoringService.h>

Public Member Functions

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

Static Public Member Functions

static void fillDescriptions (edm::ConfigurationDescriptions &descriptions)
 
static unsigned int getTID ()
 

Static Public Attributes

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

Private Member Functions

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

Static Private Member Functions

static unsigned int getSID (edm::StreamContext const &sc)
 
static unsigned int getSID (edm::StreamID const &sid)
 

Private Attributes

std::map< unsigned int, unsigned long > accuSize_
 
std::map< unsigned int, double > avgLeadTime_
 
std::unique_ptr< ConcurrencyTrackerct_
 
DAQSourcedaqInputSource_ = nullptr
 
bool exception_detected_ = false
 
std::vector< unsigned int > exceptionInLS_
 
std::string fastMicrostateDefPath_
 
unsigned int fastMonIntervals_
 
std::string fastName_
 
std::string fastPath_
 
timeval fileLookStart_
 
timeval fileLookStop_
 
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
 
std::unique_ptr< 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::vector< ContainableAtomic< const void * > > microstate_
 
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
 
std::string microstateDefPath_
 
std::string moduleLegendFile_
 
std::string moduleLegendFileJson_
 
std::atomic< bool > monInit_
 
unsigned int nMonThreads_ = 0
 
unsigned int nOutputModules_ = 0
 
unsigned int nStreams_ = 0
 
unsigned int nThreads_ = 0
 
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
 
std::filesystem::path runDirectory_
 
int sleepTime_
 
unsigned int snapCounter_ = 0
 
std::vector< std::atomic< bool > * > streamCounterUpdating_
 
bool tbbConcurrencyTracker_
 
bool tbbMonitoringMode_
 
bool threadIDAvailable_ = false
 
std::vector< ContainableAtomic< const void * > > tmicrostate_
 
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
 
std::atomic< unsigned long > totalEventsProcessed_
 
bool verbose_ = false
 
std::filesystem::path workingDirectory_
 

Detailed Description

Definition at line 164 of file FastMonitoringService.h.

Constructor & Destructor Documentation

◆ FastMonitoringService()

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

Definition at line 197 of file FastMonitoringService.cc.

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

198  : fmt_(new FastMonitoringThread()),
199  tbbMonitoringMode_(iPS.getUntrackedParameter<bool>("tbbMonitoringMode", true)),
200  tbbConcurrencyTracker_(iPS.getUntrackedParameter<bool>("tbbConcurrencyTracker", true) && tbbMonitoringMode_),
201  sleepTime_(iPS.getUntrackedParameter<int>("sleepTime", 1)),
202  fastMonIntervals_(iPS.getUntrackedParameter<unsigned int>("fastMonIntervals", 2)),
203  fastName_("fastmoni"),
205  verbose_(iPS.getUntrackedParameter<bool>("verbose")) {
206  reg.watchPreallocate(this, &FastMonitoringService::preallocate); //receiving information on number of threads
208 
213 
217 
222 
223  reg.watchPreEvent(this, &FastMonitoringService::preEvent); //stream
225 
226  //readEvent (not getNextItemType)
227  reg.watchPreSourceEvent(this, &FastMonitoringService::preSourceEvent); //source (with streamID of requestor)
229 
232 
235 
239 
240  //find microstate definition path (required by the module)
241  struct stat statbuf;
242  std::string microstateBaseSuffix = "src/EventFilter/Utilities/plugins/microstatedef.jsd";
243  std::string microstatePath = std::string(std::getenv("CMSSW_BASE")) + "/" + microstateBaseSuffix;
244  if (stat(microstatePath.c_str(), &statbuf)) {
245  microstatePath = std::string(std::getenv("CMSSW_RELEASE_BASE")) + "/" + microstateBaseSuffix;
246  if (stat(microstatePath.c_str(), &statbuf)) {
247  microstatePath = microstateBaseSuffix;
248  if (stat(microstatePath.c_str(), &statbuf))
249  throw cms::Exception("FastMonitoringService") << "microstate definition file not found";
250  }
251  }
252  fastMicrostateDefPath_ = microstateDefPath_ = microstatePath;
253  }
void watchPreStreamEarlyTermination(PreStreamEarlyTermination::slot_type const &iSlot)
void watchPreEvent(PreEvent::slot_type const &iSlot)
void postModuleEventAcquire(edm::StreamContext const &, edm::ModuleCallingContext const &)
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)
std::unique_ptr< FastMonitoringThread > fmt_
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 &)
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)
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 ( )

Definition at line 255 of file FastMonitoringService.cc.

255 {}

Member Function Documentation

◆ accumulateFileSize()

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

Definition at line 716 of file FastMonitoringService.cc.

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

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

716  {
717  std::lock_guard<std::mutex> lock(fmt_->monlock_);
718 
719  if (accuSize_.find(lumi) == accuSize_.end())
720  accuSize_[lumi] = fileSize;
721  else
722  accuSize_[lumi] += fileSize;
723 
726  else
728  }
std::map< unsigned int, unsigned long > accuSize_
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::unique_ptr< FastMonitoringThread > fmt_

◆ doSnapshot()

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

Definition at line 858 of file FastMonitoringService.cc.

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

Referenced by preGlobalEndLumi(), and snapshotRunner().

858  {
859  // update macrostate
860  fmt_->m_data.fastMacrostateJ_ = fmt_->m_data.macrostate_;
861 
862  std::vector<const void*> microstateCopy(microstate_.begin(), microstate_.end());
863  std::vector<const void*> tmicrostateCopy(tmicrostate_.begin(), tmicrostate_.end());
864  std::vector<unsigned char> microstateAcqCopy(microstateAcqFlag_.begin(), microstateAcqFlag_.end());
865  std::vector<unsigned char> tmicrostateAcqCopy(tmicrostateAcqFlag_.begin(), tmicrostateAcqFlag_.end());
866 
867  if (!isInitTransition_) {
868  auto itd = avgLeadTime_.find(ls);
869  if (itd != avgLeadTime_.end())
870  fmt_->m_data.fastAvgLeadTimeJ_ = itd->second;
871  else
872  fmt_->m_data.fastAvgLeadTimeJ_ = 0.;
873 
874  auto iti = filesProcessedDuringLumi_.find(ls);
875  if (iti != filesProcessedDuringLumi_.end())
876  fmt_->m_data.fastFilesProcessedJ_ = iti->second;
877  else
878  fmt_->m_data.fastFilesProcessedJ_ = 0;
879 
880  auto itrd = lockStatsDuringLumi_.find(ls);
881  if (itrd != lockStatsDuringLumi_.end()) {
882  fmt_->m_data.fastLockWaitJ_ = itrd->second.first;
883  fmt_->m_data.fastLockCountJ_ = itrd->second.second;
884  } else {
885  fmt_->m_data.fastLockWaitJ_ = 0.;
886  fmt_->m_data.fastLockCountJ_ = 0.;
887  }
888  }
889 
890  for (unsigned int i = 0; i < nThreads_; i++) {
891  if (tmicrostateCopy[i] == getmIdle() && ct_->isThreadActive(i)) {
892  //overhead if thread is running
893  tmicrostateCopy[i] = getmFwk();
894  }
895  if (tmicrostateAcqCopy[i])
896  fmt_->m_data.tmicrostateEncoded_[i] =
897  fmt_->m_data.microstateBins_ + fmt_->m_data.encModule_.encode(tmicrostateCopy[i]);
898  else
899  fmt_->m_data.tmicrostateEncoded_[i] = fmt_->m_data.encModule_.encode(tmicrostateCopy[i]);
900  }
901 
902  for (unsigned int i = 0; i < nStreams_; i++) {
903  if (microstateAcqCopy[i])
904  fmt_->m_data.microstateEncoded_[i] =
905  fmt_->m_data.microstateBins_ + fmt_->m_data.encModule_.encode(microstateCopy[i]);
906  else
907  fmt_->m_data.microstateEncoded_[i] = fmt_->m_data.encModule_.encode(microstateCopy[i]);
908  }
909 
910  bool inputStatePerThread = false;
911 
913  switch (inputSupervisorState_) {
915  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_fileLimit;
916  break;
918  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeChunk;
919  break;
921  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeChunkCopying;
922  break;
924  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeThread;
925  break;
927  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_waitFreeThreadCopying;
928  break;
930  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_busy;
931  break;
933  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_lockPolling;
934  break;
936  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_lockPollingCopying;
937  break;
939  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_runEnd;
940  break;
942  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_noFile;
943  break;
945  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFile;
946  break;
949  break;
951  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitThread;
952  break;
955  break;
957  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput_newFileWaitChunk;
958  break;
959  default:
960  fmt_->m_data.inputState_[0] = FastMonState::inWaitInput;
961  }
962  } else if (inputState_ == FastMonState::inWaitChunk) {
963  switch (inputSupervisorState_) {
965  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_fileLimit;
966  break;
968  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeChunk;
969  break;
971  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeChunkCopying;
972  break;
974  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeThread;
975  break;
977  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_waitFreeThreadCopying;
978  break;
980  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_busy;
981  break;
983  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_lockPolling;
984  break;
986  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_lockPollingCopying;
987  break;
989  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_runEnd;
990  break;
992  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_noFile;
993  break;
995  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFile;
996  break;
999  break;
1001  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitThread;
1002  break;
1005  break;
1007  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk_newFileWaitChunk;
1008  break;
1009  default:
1010  fmt_->m_data.inputState_[0] = FastMonState::inWaitChunk;
1011  }
1012  } else if (inputState_ == FastMonState::inNoRequest) {
1013  inputStatePerThread = true;
1014  for (unsigned int i = 0; i < nMonThreads_; i++) {
1015  if (i >= nStreams_)
1016  fmt_->m_data.inputState_[i] = FastMonState::inIgnore;
1017  else if (microstateCopy[i] == getmIdle())
1018  fmt_->m_data.inputState_[i] = FastMonState::inNoRequestWithIdleThreads;
1019  else if (microstateCopy[i] == getmEoL() || microstateCopy[i] == getmFwkEoL())
1020  fmt_->m_data.inputState_[i] = FastMonState::inNoRequestWithEoLThreads;
1021  else
1022  fmt_->m_data.inputState_[i] = FastMonState::inNoRequest;
1023  }
1024  } else if (inputState_ == FastMonState::inNewLumi) {
1025  inputStatePerThread = true;
1026  for (unsigned int i = 0; i < nMonThreads_; i++) {
1027  if (i >= nStreams_)
1028  fmt_->m_data.inputState_[i] = FastMonState::inIgnore;
1029  else if (microstateCopy[i] == getmEoL() || microstateCopy[i] == getmFwkEoL())
1030  fmt_->m_data.inputState_[i] = FastMonState::inNewLumi;
1031  }
1033  //apply directly throttled state from supervisor
1034  fmt_->m_data.inputState_[0] = inputSupervisorState_;
1035  } else
1036  fmt_->m_data.inputState_[0] = inputState_;
1037 
1038  //this is same for all streams
1039  if (!inputStatePerThread)
1040  for (unsigned int i = 1; i < nMonThreads_; i++)
1041  fmt_->m_data.inputState_[i] = fmt_->m_data.inputState_[0];
1042 
1043  if (isGlobalEOL) { //only update global variables
1044  fmt_->jsonMonitor_->snapGlobal(ls);
1045  } else
1046  fmt_->jsonMonitor_->snap(ls);
1047  }
constexpr edm::ModuleDescription const * getmFwkEoL()
std::atomic< FastMonState::InputState > inputState_
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
std::atomic< bool > isInitTransition_
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::unique_ptr< FastMonitoringThread > fmt_
constexpr edm::ModuleDescription const * getmFwk()
constexpr edm::ModuleDescription const * getmEoL()
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
std::map< unsigned int, double > avgLeadTime_
std::atomic< FastMonState::InputState > inputSupervisorState_
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
std::unique_ptr< ConcurrencyTracker > ct_
def ls(path, rec=False)
Definition: eostools.py:349
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
if(threadIdxLocalY==0 &&threadIdxLocalX==0)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ exceptionDetected()

bool evf::FastMonitoringService::exceptionDetected ( ) const

Definition at line 456 of file FastMonitoringService.cc.

References has_data_exception_, and has_source_exception_.

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

456  {
457  return has_source_exception_.load() || has_data_exception_.load();
458  }
std::atomic< bool > has_data_exception_
std::atomic< bool > has_source_exception_

◆ fillDescriptions()

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

Definition at line 257 of file FastMonitoringService.cc.

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

257  {
259  desc.setComment("Service for File-based DAQ monitoring and event accounting");
260  desc.addUntracked<bool>("tbbMonitoringMode", true)
261  ->setComment("Monitor individual module processing per TBB thread instead of stream");
262  desc.addUntracked<bool>("tbbConcurrencyTracker", true)
263  ->setComment("Monitor TBB thread activity to flag microstate as real idle or overhead/other");
264  desc.addUntracked<int>("sleepTime", 1)->setComment("Sleep time of the monitoring thread");
265  desc.addUntracked<unsigned int>("fastMonIntervals", 2)
266  ->setComment("Modulo of sleepTime intervals on which fastmon file is written out");
267  desc.addUntracked<bool>("filePerFwkStream", true) //obsolete
268  ->setComment("Switches on monitoring output per framework stream");
269  desc.addUntracked<bool>("verbose", false)->setComment("Set to use LogInfo messages from the monitoring thread");
270  desc.setAllowAnything();
271  descriptions.add("FastMonitoringService", desc);
272  }
void add(std::string const &label, ParameterSetDescription const &psetDescription)

◆ getAbortFlagForLumi()

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

Definition at line 794 of file FastMonitoringService.cc.

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

Referenced by shouldWriteFiles().

794  {
795  std::lock_guard<std::mutex> lock(fmt_->monlock_);
796 
797  auto it = processedEventsPerLumi_.find(lumi);
798  if (it != processedEventsPerLumi_.end()) {
799  unsigned int abortFlag = it->second.second;
800  return abortFlag;
801  } else {
802  throw cms::Exception("FastMonitoringService")
803  << "output module wants already deleted (or never reported by SOURCE) lumisection status for LUMI -: "
804  << lumi;
805  return false;
806  }
807  }
std::unique_ptr< FastMonitoringThread > fmt_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_

◆ getEventsProcessedForLumi()

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

Definition at line 776 of file FastMonitoringService.cc.

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

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

776  {
777  std::lock_guard<std::mutex> lock(fmt_->monlock_);
778 
779  auto it = processedEventsPerLumi_.find(lumi);
780  if (it != processedEventsPerLumi_.end()) {
781  unsigned int proc = it->second.first;
782  if (abortFlag)
783  *abortFlag = it->second.second;
784  return proc;
785  } else {
786  throw cms::Exception("FastMonitoringService")
787  << "output module wants already deleted (or never reported by SOURCE) lumisection event count for LUMI -: "
788  << lumi;
789  return 0;
790  }
791  }
std::unique_ptr< FastMonitoringThread > fmt_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_

◆ getRunDirName()

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

Definition at line 223 of file FastMonitoringService.h.

References runDirectory_.

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

◆ getSID() [1/2]

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

◆ getSID() [2/2]

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

Definition at line 240 of file FastMonitoringService.h.

References edm::StreamID::value().

240 { return sid.value(); }

◆ getTID()

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

Definition at line 231 of file FastMonitoringService.h.

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

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

◆ isExceptionOnData()

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

Definition at line 460 of file FastMonitoringService.cc.

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

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

460  {
461  if (!has_data_exception_.load())
462  return false;
463  if (has_source_exception_.load())
464  return true;
465  std::lock_guard<std::mutex> lock(fmt_->monlock_);
466  for (auto ex : exceptionInLS_) {
467  if (ls == ex)
468  return true;
469  }
470  return false;
471  }
std::atomic< bool > has_data_exception_
std::unique_ptr< FastMonitoringThread > fmt_
def ls(path, rec=False)
Definition: eostools.py:349
std::atomic< bool > has_source_exception_
std::vector< unsigned int > exceptionInLS_

◆ jobFailure()

void evf::FastMonitoringService::jobFailure ( )

Definition at line 473 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

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

◆ makeInputLegendaJson()

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

Definition at line 295 of file FastMonitoringService.cc.

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

Referenced by postBeginJob().

295  {
296  Json::Value legendaVector(Json::arrayValue);
297  for (int i = 0; i < FastMonState::inCOUNT; i++)
298  legendaVector.append(Json::Value(inputStateNames[i]));
299  Json::Value moduleLegend;
300  moduleLegend["names"] = legendaVector;
302  return writer.write(moduleLegend);
303  }
static const std::string inputStateNames[FastMonState::inCOUNT]
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:64
array value (ordered list)
Definition: value.h:32
Represents a JSON value.
Definition: value.h:101

◆ makeModuleLegendaJson()

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

Definition at line 274 of file FastMonitoringService.cc.

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

Referenced by postBeginJob().

274  {
275  Json::Value legendaVector(Json::arrayValue);
276  for (int i = 0; i < fmt_->m_data.encModule_.current_; i++)
277  legendaVector.append(
278  Json::Value((static_cast<const edm::ModuleDescription*>(fmt_->m_data.encModule_.decode(i)))->moduleLabel()));
279  //duplicate modules adding a list for acquire states (not all modules actually have it)
280  for (int i = 0; i < fmt_->m_data.encModule_.current_; i++)
281  legendaVector.append(Json::Value(
282  (static_cast<const edm::ModuleDescription*>(fmt_->m_data.encModule_.decode(i)))->moduleLabel() + "__ACQ"));
283  Json::Value valReserved(nReservedModules);
284  Json::Value valSpecial(nSpecialModules);
285  Json::Value valOutputModules(nOutputModules_);
286  Json::Value moduleLegend;
287  moduleLegend["names"] = legendaVector;
288  moduleLegend["reserved"] = valReserved;
289  moduleLegend["special"] = valSpecial;
290  moduleLegend["output"] = valOutputModules;
292  return writer.write(moduleLegend);
293  }
std::unique_ptr< FastMonitoringThread > fmt_
constexpr int nSpecialModules
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:64
array value (ordered list)
Definition: value.h:32
constexpr int nReservedModules
Represents a JSON value.
Definition: value.h:101

◆ postBeginJob()

void evf::FastMonitoringService::postBeginJob ( )

Definition at line 490 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

490  {
491  std::string&& moduleLegStrJson = makeModuleLegendaJson();
492  FileIO::writeStringToFile(moduleLegendFileJson_, moduleLegStrJson);
493 
494  std::string inputLegendStrJson = makeInputLegendaJson();
495  FileIO::writeStringToFile(inputLegendFileJson_, inputLegendStrJson);
496 
497  fmt_->m_data.macrostate_ = FastMonState::sJobReady;
498 
499  //update number of entries in module histogram
500  std::lock_guard<std::mutex> lock(fmt_->monlock_);
501  //double the size to add post-acquire states
502  fmt_->m_data.microstateBins_ = fmt_->m_data.encModule_.vecsize() * 2;
503  }
std::unique_ptr< FastMonitoringThread > fmt_

◆ postEndJob()

void evf::FastMonitoringService::postEndJob ( )

Definition at line 505 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

505  {
506  fmt_->m_data.macrostate_ = FastMonState::sJobEnded;
507  fmt_->stop();
508  }
std::unique_ptr< FastMonitoringThread > fmt_

◆ postEvent()

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

Definition at line 640 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

640  {
641  (*(fmt_->m_data.processed_[sc.streamID().value()]))++;
642  //fast path counter (events accumulated in a run)
643  unsigned long res = totalEventsProcessed_.fetch_add(1, std::memory_order_relaxed);
644  fmt_->m_data.fastPathProcessedJ_ = res + 1;
645 
646  microstate_[sc.streamID().value()] = getmIdle();
647  }
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
std::unique_ptr< FastMonitoringThread > fmt_
Definition: Electron.h:6
std::atomic< unsigned long > totalEventsProcessed_

◆ postGlobalBeginRun()

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

Definition at line 510 of file FastMonitoringService.cc.

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

510  {
511  fmt_->m_data.macrostate_ = FastMonState::sRunning;
512  isInitTransition_ = false;
513  }
std::atomic< bool > isInitTransition_
std::unique_ptr< FastMonitoringThread > fmt_

◆ postGlobalEndLumi()

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

Definition at line 597 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

597  {
598  std::lock_guard<std::mutex> lock(fmt_->monlock_);
599  unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
600  //LS monitoring snapshot with input source data has been taken in previous callback
601  avgLeadTime_.erase(lumi);
603  lockStatsDuringLumi_.erase(lumi);
604 
605  //output module already used this in end lumi (this could be migrated to EvFDaqDirector as it is essential for FFF bookkeeping)
607  }
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::unique_ptr< FastMonitoringThread > fmt_
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_

◆ postModuleEvent()

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

Definition at line 705 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

705  {
707  if (!tbbMonitoringMode_)
708  return;
709  auto tid = getTID();
710  if (tid >= nThreads_)
711  return;
712  tmicrostate_[tid] = getmIdle();
713  }
constexpr edm::ModuleDescription const * getmFwkOvhMod()
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ postModuleEventAcquire()

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

Definition at line 682 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

683  {
685  microstateAcqFlag_[getSID(sc)] = 0;
686  if (!tbbMonitoringMode_)
687  return;
688  auto tid = getTID();
689  if (tid >= nThreads_)
690  return;
691  tmicrostate_[tid] = getmIdle();
692  tmicrostateAcqFlag_[tid] = 0;
693  }
constexpr edm::ModuleDescription const * getmFwkOvhMod()
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ postSourceEvent()

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

Definition at line 659 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

659  {
660  microstate_[getSID(sid)] = getmFwkOvhSrc();
661  if (!tbbMonitoringMode_)
662  return;
663  auto tid = getTID();
664  if (tid >= nThreads_)
665  return;
666  tmicrostate_[tid] = getmIdle();
667  }
constexpr edm::ModuleDescription const * getmFwkOvhSrc()
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ postStreamBeginLumi()

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

Definition at line 619 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

619  {
620  microstate_[sc.streamID().value()] = getmIdle();
621  }
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmIdle()

◆ postStreamEndLumi()

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

Definition at line 632 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

632  {
633  microstate_[sc.streamID().value()] = getmFwkEoL();
634  }
constexpr edm::ModuleDescription const * getmFwkEoL()
std::vector< ContainableAtomic< const void * > > microstate_

◆ preallocate()

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

Definition at line 305 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

305  {
306  nStreams_ = bounds.maxNumberOfStreams();
307  nThreads_ = bounds.maxNumberOfThreads();
308  //this should already be >=1
309  if (nStreams_ == 0)
310  nStreams_ = 1;
311  if (nThreads_ == 0)
312  nThreads_ = 1;
314  ct_ = std::make_unique<ConcurrencyTracker>(nThreads_);
315  //start concurrency tracking
316  }
std::unique_ptr< ConcurrencyTracker > ct_

◆ preBeginJob()

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

Definition at line 318 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

318  {
319  // FIND RUN DIRECTORY
320  // The run dir should be set via the configuration of EvFDaqDirector
322  ct_->activate();
323 
324  if (edm::Service<evf::EvFDaqDirector>().operator->() == nullptr) {
325  throw cms::Exception("FastMonitoringService") << "EvFDaqDirector is not present";
326  }
327  std::filesystem::path runDirectory{edm::Service<evf::EvFDaqDirector>()->baseRunDir()};
328  workingDirectory_ = runDirectory_ = runDirectory;
329  workingDirectory_ /= "mon";
330 
331  if (!std::filesystem::is_directory(workingDirectory_)) {
332  LogDebug("FastMonitoringService") << "<MON> DIR NOT FOUND! Trying to create -: " << workingDirectory_.string();
333  std::filesystem::create_directories(workingDirectory_);
334  if (!std::filesystem::is_directory(workingDirectory_))
335  edm::LogWarning("FastMonitoringService") << "Unable to create <MON> DIR -: " << workingDirectory_.string()
336  << ". No monitoring data will be written.";
337  }
338 
339  std::ostringstream fastFileName;
340 
341  fastFileName << fastName_ << "_pid" << std::setfill('0') << std::setw(5) << getpid() << ".fast";
343  fast /= fastFileName.str();
344  fastPath_ = fast.string();
345 
346  std::ostringstream moduleLegFile;
347  std::ostringstream moduleLegFileJson;
348  moduleLegFile << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
349  moduleLegFileJson << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
350  moduleLegendFile_ = (workingDirectory_ / moduleLegFile.str()).string();
351  moduleLegendFileJson_ = (workingDirectory_ / moduleLegFileJson.str()).string();
352 
353  std::ostringstream inputLegFileJson;
354  inputLegFileJson << "inputlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
355  inputLegendFileJson_ = (workingDirectory_ / inputLegFileJson.str()).string();
356 
357  LogDebug("FastMonitoringService") << "Initializing FastMonitor with microstate def path -: " << microstateDefPath_;
358 
359  /*
360  * initialize the fast monitor with:
361  * vector of pointers to monitorable parameters
362  * path to definition
363  *
364  */
365 
366  fmt_->m_data.macrostate_ = FastMonState::sInit;
367 
368  for (unsigned int i = 0; i < (FastMonState::mCOUNT); i++)
369  fmt_->m_data.encModule_.updateReserved(static_cast<const void*>(specialMicroStateNames + i));
370  fmt_->m_data.encModule_.completeReservedWithDummies();
371 
372  for (unsigned int i = 0; i < nMonThreads_; i++) {
373  microstate_.emplace_back(getmInvalid());
374  microstateAcqFlag_.push_back(0);
375  tmicrostate_.emplace_back(getmInvalid());
376  tmicrostateAcqFlag_.push_back(0);
377 
378  //for synchronization
379  streamCounterUpdating_.push_back(new std::atomic<bool>(false));
380  }
381 
382  //initial size until we detect number of bins
383  fmt_->m_data.macrostateBins_ = FastMonState::MCOUNT;
384  fmt_->m_data.microstateBins_ = 0;
385  fmt_->m_data.inputstateBins_ = FastMonState::inCOUNT;
386 
387  lastGlobalLumi_ = 0;
388  isInitTransition_ = true;
389  lumiFromSource_ = 0;
390 
391  //startup monitoring
392  fmt_->resetFastMonitor(microstateDefPath_, fastMicrostateDefPath_);
393  fmt_->jsonMonitor_->setNStreams(nMonThreads_);
394  fmt_->m_data.registerVariables(fmt_->jsonMonitor_.get(), nMonThreads_, nStreams_, nThreads_);
395  monInit_.store(false, std::memory_order_release);
396  if (sleepTime_ > 0)
398  }
constexpr edm::ModuleDescription const * getmInvalid()
std::vector< ContainableAtomic< const void * > > microstate_
std::atomic< bool > isInitTransition_
std::filesystem::path workingDirectory_
std::vector< std::atomic< bool > * > streamCounterUpdating_
std::unique_ptr< FastMonitoringThread > fmt_
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
std::filesystem::path runDirectory_
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
std::unique_ptr< ConcurrencyTracker > ct_
Log< level::Warning, false > LogWarning
static const edm::ModuleDescription specialMicroStateNames[FastMonState::mCOUNT]
#define LogDebug(id)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ preEvent()

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

Definition at line 636 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

636  {
637  microstate_[sc.streamID().value()] = getmEvent();
638  }
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmEvent()

◆ preGlobalBeginLumi()

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

Definition at line 515 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

515  {
516  timeval lumiStartTime;
517  gettimeofday(&lumiStartTime, nullptr);
518  unsigned int newLumi = gc.luminosityBlockID().luminosityBlock();
519  lastGlobalLumi_ = newLumi;
520 
521  std::lock_guard<std::mutex> lock(fmt_->monlock_);
522  lumiStartTime_[newLumi] = lumiStartTime;
523  //reset all states to idle
524  if (tbbMonitoringMode_)
525  for (unsigned i = 0; i < nThreads_; i++)
526  if (tmicrostate_[i] == getmInvalid())
527  tmicrostate_[i] = getmIdle();
528  }
constexpr edm::ModuleDescription const * getmInvalid()
constexpr edm::ModuleDescription const * getmIdle()
std::map< unsigned int, timeval > lumiStartTime_
std::unique_ptr< FastMonitoringThread > fmt_
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ preGlobalEarlyTermination()

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

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

416  {
419  context = " FromThisContext ";
421  context = " FromAnotherContext";
423  context = " FromExternalSignal";
424  edm::LogWarning("FastMonitoringService")
425  << " GLOBAL "
426  << "earlyTermination -: LS:" << gc.luminosityBlockID().luminosityBlock() << " " << context;
427  std::lock_guard<std::mutex> lock(fmt_->monlock_);
428  exceptionInLS_.push_back(gc.luminosityBlockID().luminosityBlock());
429  has_data_exception_.store(true);
430  }
std::atomic< bool > has_data_exception_
std::unique_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 530 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

530  {
531  unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
532  LogDebug("FastMonitoringService") << "Lumi ended. Writing JSON information. LUMI -: " << lumi;
533  timeval lumiStopTime;
534  gettimeofday(&lumiStopTime, nullptr);
535 
536  std::lock_guard<std::mutex> lock(fmt_->monlock_);
537 
538  // Compute throughput
539  timeval stt = lumiStartTime_[lumi];
540  lumiStartTime_.erase(lumi);
541  unsigned long usecondsForLumi = (lumiStopTime.tv_sec - stt.tv_sec) * 1000000 + (lumiStopTime.tv_usec - stt.tv_usec);
542  unsigned long accuSize = accuSize_.find(lumi) == accuSize_.end() ? 0 : accuSize_[lumi];
543  accuSize_.erase(lumi);
544  double throughput = throughputFactor() * double(accuSize) / double(usecondsForLumi);
545  //store to registered variable
546  fmt_->m_data.fastThroughputJ_.value() = throughput;
547 
548  //update
549  doSnapshot(lumi, true);
550 
551  //retrieve one result we need (todo: sanity check if it's found)
552  IntJ* lumiProcessedJptr = dynamic_cast<IntJ*>(fmt_->jsonMonitor_->getMergedIntJForLumi("Processed", lumi));
553  if (!lumiProcessedJptr)
554  throw cms::Exception("FastMonitoringService") << "Internal error: got null pointer from FastMonitor";
555  processedEventsPerLumi_[lumi] = std::pair<unsigned int, bool>(lumiProcessedJptr->value(), false);
556 
557  //checking if exception has been thrown (in case of Global/Stream early termination, for this LS)
558  bool exception_detected = exception_detected_;
559  for (auto ex : exceptionInLS_)
560  if (lumi == ex)
561  exception_detected = true;
562 
563  if (edm::shutdown_flag || exception_detected) {
564  edm::LogInfo("FastMonitoringService")
565  << "Run interrupted. Skip writing EoL information -: " << processedEventsPerLumi_[lumi].first
566  << " events were processed in LUMI " << lumi;
567  //this will prevent output modules from producing json file for possibly incomplete lumi
568  processedEventsPerLumi_[lumi].first = 0;
569  processedEventsPerLumi_[lumi].second = true;
570  //disable this exception, so service can be used standalone (will be thrown if output module asks for this information)
571  //throw cms::Exception("FastMonitoringService") << "SOURCE did not send update for lumi block. LUMI -:" << lumi;
572  return;
573  }
574 
575  if (inputSource_ || daqInputSource_) {
576  auto sourceReport =
578  if (sourceReport.first) {
579  if (sourceReport.second != processedEventsPerLumi_[lumi].first) {
580  throw cms::Exception("FastMonitoringService") << "MISMATCH with SOURCE update. LUMI -: " << lumi
581  << ", events(processed):" << processedEventsPerLumi_[lumi].first
582  << " events(source):" << sourceReport.second;
583  }
584  }
585  }
586 
587  edm::LogInfo("FastMonitoringService")
588  << "Statistics for lumisection -: lumi = " << lumi << " events = " << lumiProcessedJptr->value()
589  << " time = " << usecondsForLumi / 1000000 << " size = " << accuSize << " thr = " << throughput;
590  delete lumiProcessedJptr;
591 
592  //full global and stream merge (will be used by output modules), output from this service is deprecated
593  fmt_->jsonMonitor_->outputFullJSON("dummy", lumi, false);
594  fmt_->jsonMonitor_->discardCollected(lumi); //we don't do further updates for this lumi
595  }
std::map< unsigned int, timeval > lumiStartTime_
std::pair< bool, unsigned int > getEventReport(unsigned int lumi, bool erase)
std::map< unsigned int, unsigned long > accuSize_
volatile std::atomic< bool > shutdown_flag
std::pair< bool, unsigned int > getEventReport(unsigned int lumi, bool erase)
Definition: DAQSource.cc:1390
std::unique_ptr< FastMonitoringThread > fmt_
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_
std::vector< unsigned int > exceptionInLS_
#define LogDebug(id)

◆ preModuleBeginJob()

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

Definition at line 476 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

476  {
477  std::lock_guard<std::mutex> lock(fmt_->monlock_);
478  //std::cout << " Pre module Begin Job module: " << desc.moduleName() << std::endl;
479 
480  //build a map of modules keyed by their module description address
481  //here we need to treat output modules in a special way so they can be easily singled out
482  if (desc.moduleName() == "Stream" || desc.moduleName() == "GlobalEvFOutputModule" ||
483  desc.moduleName() == "EventStreamFileWriter" || desc.moduleName() == "PoolOutputModule") {
484  fmt_->m_data.encModule_.updateReserved((void*)&desc);
485  nOutputModules_++;
486  } else
487  fmt_->m_data.encModule_.update((void*)&desc);
488  }
std::unique_ptr< FastMonitoringThread > fmt_

◆ preModuleEvent()

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

Definition at line 695 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

695  {
696  microstate_[getSID(sc)] = (void*)(mcc.moduleDescription());
697  if (!tbbMonitoringMode_)
698  return;
699  auto tid = getTID();
700  if (tid >= nThreads_)
701  return;
702  tmicrostate_[tid] = (void*)(mcc.moduleDescription());
703  }
std::vector< ContainableAtomic< const void * > > microstate_
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ preModuleEventAcquire()

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

Definition at line 669 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

670  {
671  microstate_[getSID(sc)] = (void*)(mcc.moduleDescription());
672  microstateAcqFlag_[getSID(sc)] = 1;
673  if (!tbbMonitoringMode_)
674  return;
675  auto tid = getTID();
676  if (tid >= nThreads_)
677  return;
678  tmicrostate_[tid] = (void*)(mcc.moduleDescription());
679  tmicrostateAcqFlag_[tid] = 1;
680  }
std::vector< ContainableAtomic< const void * > > microstate_
std::vector< ContainableAtomic< unsigned char > > tmicrostateAcqFlag_
std::vector< ContainableAtomic< unsigned char > > microstateAcqFlag_
static unsigned int getSID(edm::StreamContext const &sc)
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ prePathEvent()

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

◆ preSourceEarlyTermination()

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

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

432  {
435  context = " FromThisContext ";
437  context = " FromAnotherContext";
439  context = " FromExternalSignal";
440  edm::LogWarning("FastMonitoringService") << " SOURCE "
441  << "earlyTermination -: " << context;
442  std::lock_guard<std::mutex> lock(fmt_->monlock_);
443  exception_detected_ = true;
444  has_source_exception_.store(true);
445  has_data_exception_.store(true);
446  }
std::atomic< bool > has_data_exception_
std::unique_ptr< FastMonitoringThread > fmt_
std::atomic< bool > has_source_exception_
Log< level::Warning, false > LogWarning

◆ preSourceEvent()

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

Definition at line 649 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

649  {
650  microstate_[getSID(sid)] = getmInput();
651  if (!tbbMonitoringMode_)
652  return;
653  auto tid = getTID();
654  if (tid >= nThreads_)
655  return;
656  tmicrostate_[tid] = getmInput();
657  }
std::vector< ContainableAtomic< const void * > > microstate_
static unsigned int getSID(edm::StreamContext const &sc)
constexpr edm::ModuleDescription const * getmInput()
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ preStreamBeginLumi()

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

Definition at line 609 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

609  {
610  std::lock_guard<std::mutex> lock(fmt_->monlock_);
611  fmt_->m_data.streamLumi_[sc.streamID().value()] = sc.eventID().luminosityBlock();
612 
613  //reset collected values for this stream
614  *(fmt_->m_data.processed_[sc.streamID().value()]) = 0;
615 
616  microstate_[sc.streamID().value()] = getmBoL();
617  }
std::vector< ContainableAtomic< const void * > > microstate_
constexpr edm::ModuleDescription const * getmBoL()
std::unique_ptr< FastMonitoringThread > fmt_

◆ preStreamEarlyTermination()

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

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

400  {
403  context = " FromThisContext ";
405  context = " FromAnotherContext";
407  context = " FromExternalSignal";
408  edm::LogWarning("FastMonitoringService")
409  << " STREAM " << sc.streamID().value() << " earlyTermination -: ID:" << sc.eventID()
410  << " LS:" << sc.eventID().luminosityBlock() << " " << context;
411  std::lock_guard<std::mutex> lock(fmt_->monlock_);
412  exceptionInLS_.push_back(sc.eventID().luminosityBlock());
413  has_data_exception_.store(true);
414  }
std::atomic< bool > has_data_exception_
std::unique_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 623 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

623  {
624  std::lock_guard<std::mutex> lock(fmt_->monlock_);
625  //update processed count to be complete at this time
626  //doStreamEOLSnapshot(sc.eventID().luminosityBlock(), sid);
627  fmt_->jsonMonitor_->snapStreamAtomic(sc.eventID().luminosityBlock(), sc.streamID().value());
628  //reset this in case stream does not get notified of next lumi (we keep processed events only)
629  microstate_[sc.streamID().value()] = getmEoL();
630  }
std::vector< ContainableAtomic< const void * > > microstate_
std::unique_ptr< FastMonitoringThread > fmt_
constexpr edm::ModuleDescription const * getmEoL()

◆ reportLockWait()

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

Definition at line 766 of file FastMonitoringService.cc.

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

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

766  {
767  std::lock_guard<std::mutex> lock(fmt_->monlock_);
768  lockStatsDuringLumi_[ls] = std::pair<double, unsigned int>(waitTime, lockCount);
769  }
std::unique_ptr< FastMonitoringThread > fmt_
def ls(path, rec=False)
Definition: eostools.py:349
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_

◆ setExceptionDetected()

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

Definition at line 448 of file FastMonitoringService.cc.

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

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

448  {
449  std::lock_guard<std::mutex> lock(fmt_->monlock_);
450  if (!ls)
451  exception_detected_ = true;
452  else
453  exceptionInLS_.push_back(ls);
454  }
std::unique_ptr< FastMonitoringThread > fmt_
def ls(path, rec=False)
Definition: eostools.py:349
std::vector< unsigned int > exceptionInLS_

◆ setInputSource() [1/2]

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

Definition at line 224 of file FastMonitoringService.h.

References inputSource_.

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

224 { inputSource_ = inputSource; }
FedRawDataInputSource * inputSource_

◆ setInputSource() [2/2]

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

Definition at line 225 of file FastMonitoringService.h.

References daqInputSource_.

225 { daqInputSource_ = inputSource; }

◆ setInState()

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

◆ setInStateSup()

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

◆ setTMicrostate()

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

Definition at line 771 of file FastMonitoringService.cc.

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

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

771  {
772  tmicrostate_[tbb::this_task_arena::current_thread_index()] = &specialMicroStateNames[m];
773  }
static const edm::ModuleDescription specialMicroStateNames[FastMonState::mCOUNT]
std::vector< ContainableAtomic< const void * > > tmicrostate_

◆ shouldWriteFiles()

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

Definition at line 217 of file FastMonitoringService.h.

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

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

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

◆ snapshotRunner()

void evf::FastMonitoringService::snapshotRunner ( )
private

Definition at line 810 of file FastMonitoringService.cc.

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

Referenced by preBeginJob().

810  {
811  monInit_.exchange(true, std::memory_order_acquire);
812  while (!fmt_->m_stoprequest) {
813  std::vector<std::vector<unsigned int>> lastEnc;
814  {
815  std::unique_lock<std::mutex> lock(fmt_->monlock_);
816 
817  doSnapshot(lastGlobalLumi_, false);
818 
819  lastEnc.emplace_back(fmt_->m_data.tmicrostateEncoded_);
820  lastEnc.emplace_back(fmt_->m_data.microstateEncoded_);
821 
823  std::vector<std::string> CSVv;
824  for (unsigned int i = 0; i < nMonThreads_; i++) {
825  CSVv.push_back(fmt_->jsonMonitor_->getCSVString((int)i));
826  }
827  // release mutex before writing out fast path file
828  lock.release()->unlock();
829  fmt_->jsonMonitor_->outputCSV(fastPath_, CSVv);
830  }
831  snapCounter_++;
832  }
833 
834  if (verbose_) {
835  edm::LogInfo msg("FastMonitoringService");
836  auto f = [&](std::vector<unsigned int> const& p) {
837  for (unsigned int i = 0; i < nMonThreads_; i++) {
838  if (i == 0)
839  msg << "[" << p[i] << ",";
840  else if (i <= nMonThreads_ - 1)
841  msg << p[i] << ",";
842  else
843  msg << p[i] << "]";
844  }
845  };
846 
847  msg << "Current states: Ms=" << fmt_->m_data.fastMacrostateJ_.value() << " ms=";
848  f(lastEnc[0]);
849  msg << " us=";
850  f(lastEnc[1]);
852  }
853 
854  ::sleep(sleepTime_);
855  }
856  }
std::atomic< FastMonState::InputState > inputState_
static const std::string inputStateNames[FastMonState::inCOUNT]
std::unique_ptr< FastMonitoringThread > fmt_
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
std::atomic< FastMonState::InputState > inputSupervisorState_
double f[11][100]
tuple msg
Definition: mps_check.py:286

◆ startedLookingForFile()

void evf::FastMonitoringService::startedLookingForFile ( )

Definition at line 730 of file FastMonitoringService.cc.

References fileLookStart_.

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

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

◆ stoppedLookingForFile()

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

Definition at line 738 of file FastMonitoringService.cc.

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

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

738  {
739  gettimeofday(&fileLookStop_, nullptr);
740  /*
741  std::cout << "Stopped looking for .raw file at: s=" << fileLookStop_.tv_sec << ": ms = "
742  << fileLookStop_.tv_usec / 1000.0 << std::endl;
743  */
744  std::lock_guard<std::mutex> lock(fmt_->monlock_);
745 
746  if (lumi > lumiFromSource_) {
748  leadTimes_.clear();
749  }
750  unsigned long elapsedTime = (fileLookStop_.tv_sec - fileLookStart_.tv_sec) * 1000000 // sec to us
751  + (fileLookStop_.tv_usec - fileLookStart_.tv_usec); // us
752  // add this to lead times for this lumi
753  leadTimes_.push_back((double)elapsedTime);
754 
755  // recompute average lead time for this lumi
756  if (leadTimes_.size() == 1)
758  else {
759  double totTime = 0;
760  for (unsigned int i = 0; i < leadTimes_.size(); i++)
761  totTime += leadTimes_[i];
762  avgLeadTime_[lumi] = 0.001 * (totTime / leadTimes_.size());
763  }
764  }
std::unique_ptr< FastMonitoringThread > fmt_
std::map< unsigned int, double > avgLeadTime_
std::vector< double > leadTimes_

Member Data Documentation

◆ accuSize_

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

Definition at line 277 of file FastMonitoringService.h.

Referenced by accumulateFileSize(), and preGlobalEndLumi().

◆ avgLeadTime_

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

Definition at line 274 of file FastMonitoringService.h.

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

◆ ct_

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

Definition at line 244 of file FastMonitoringService.h.

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

◆ daqInputSource_

DAQSource* evf::FastMonitoringService::daqInputSource_ = nullptr
private

Definition at line 248 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

◆ 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 260 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

◆ fastMonIntervals_

unsigned int evf::FastMonitoringService::fastMonIntervals_
private

Definition at line 258 of file FastMonitoringService.h.

Referenced by snapshotRunner().

◆ fastName_

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

Definition at line 261 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ fastPath_

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

Definition at line 261 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ fileLookStart_

timeval evf::FastMonitoringService::fileLookStart_
private

Definition at line 266 of file FastMonitoringService.h.

Referenced by startedLookingForFile(), and stoppedLookingForFile().

◆ fileLookStop_

timeval evf::FastMonitoringService::fileLookStop_
private

Definition at line 266 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

◆ filesProcessedDuringLumi_

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

Definition at line 275 of file FastMonitoringService.h.

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

◆ fmt_

std::unique_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 296 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

◆ inputSource_

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

Definition at line 247 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and setInputSource().

◆ inputState_

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

Definition at line 249 of file FastMonitoringService.h.

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

◆ inputStateNames

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

Definition at line 169 of file FastMonitoringService.h.

Referenced by makeInputLegendaJson(), and snapshotRunner().

◆ inputSupervisorState_

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

Definition at line 250 of file FastMonitoringService.h.

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

◆ isInitTransition_

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

Definition at line 269 of file FastMonitoringService.h.

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

◆ lastGlobalLumi_

unsigned int evf::FastMonitoringService::lastGlobalLumi_
private

Definition at line 268 of file FastMonitoringService.h.

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

◆ leadTimes_

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

Definition at line 278 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

◆ lockStatsDuringLumi_

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

Definition at line 279 of file FastMonitoringService.h.

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

◆ lumiFromSource_

unsigned int evf::FastMonitoringService::lumiFromSource_
private

Definition at line 270 of file FastMonitoringService.h.

Referenced by preBeginJob(), and stoppedLookingForFile().

◆ lumiStartTime_

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

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

◆ microstate_

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

◆ microstateAcqFlag_

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

◆ microstateDefPath_

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

Definition at line 260 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preBeginJob().

◆ moduleLegendFile_

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

Definition at line 294 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ moduleLegendFileJson_

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

Definition at line 295 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preBeginJob().

◆ monInit_

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

Definition at line 299 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ nMonThreads_

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

Definition at line 253 of file FastMonitoringService.h.

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

◆ nOutputModules_

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

Definition at line 297 of file FastMonitoringService.h.

Referenced by makeModuleLegendaJson(), and preModuleBeginJob().

◆ nStreams_

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

Definition at line 252 of file FastMonitoringService.h.

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

◆ nThreads_

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

◆ processedEventsPerLumi_

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

◆ runDirectory_

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

Definition at line 288 of file FastMonitoringService.h.

Referenced by getRunDirName(), and preBeginJob().

◆ sleepTime_

int evf::FastMonitoringService::sleepTime_
private

Definition at line 257 of file FastMonitoringService.h.

Referenced by preBeginJob(), and snapshotRunner().

◆ snapCounter_

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

Definition at line 259 of file FastMonitoringService.h.

Referenced by snapshotRunner().

◆ specialMicroStateNames

const edm::ModuleDescription evf::FastMonitoringService::specialMicroStateNames
static
Initial value:
= {
edm::ModuleDescription("Dummy", "Invalid"),
edm::ModuleDescription("Dummy", "Idle"),
edm::ModuleDescription("Dummy", "FwkOvhSrc"),
edm::ModuleDescription("Dummy", "FwkOvhMod"),
edm::ModuleDescription("Dummy", "FwkEoL"),
edm::ModuleDescription("Dummy", "Input"),
edm::ModuleDescription("Dummy", "DQM"),
edm::ModuleDescription("Dummy", "BoL"),
edm::ModuleDescription("Dummy", "EoL"),
edm::ModuleDescription("Dummy", "GlobalEoL"),
edm::ModuleDescription("Dummy", "Fwk"),
edm::ModuleDescription("Dummy", "IdleSource"),
edm::ModuleDescription("Dummy", "Event"),
edm::ModuleDescription("Dummy", "Ignore")}

Definition at line 167 of file FastMonitoringService.h.

Referenced by preBeginJob(), and setTMicrostate().

◆ streamCounterUpdating_

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

Definition at line 286 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ tbbConcurrencyTracker_

bool evf::FastMonitoringService::tbbConcurrencyTracker_
private

Definition at line 256 of file FastMonitoringService.h.

Referenced by preBeginJob().

◆ tbbMonitoringMode_

bool evf::FastMonitoringService::tbbMonitoringMode_
private

◆ threadIDAvailable_

bool evf::FastMonitoringService::threadIDAvailable_ = false
private

Definition at line 290 of file FastMonitoringService.h.

◆ tmicrostate_

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

◆ tmicrostateAcqFlag_

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

◆ totalEventsProcessed_

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

Definition at line 292 of file FastMonitoringService.h.

Referenced by postEvent().

◆ verbose_

bool evf::FastMonitoringService::verbose_ = false
private

Definition at line 312 of file FastMonitoringService.h.

Referenced by snapshotRunner().

◆ workingDirectory_

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

Definition at line 288 of file FastMonitoringService.h.

Referenced by preBeginJob().