CMS 3D CMS Logo

 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Pages
List of all members | Classes | 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

Classes

struct  Encoding
 

Public Member Functions

void accumulateFileSize (unsigned int lumi, unsigned long fileSize)
 
 FastMonitoringService (const edm::ParameterSet &, edm::ActivityRegistry &)
 
bool getAbortFlagForLumi (unsigned int lumi)
 
unsigned int getEventsProcessedForLumi (unsigned int lumi, bool *abortFlag=0)
 
std::string getRunDirName () const
 
void jobFailure ()
 
std::string makeModuleLegendaJson ()
 
std::string makePathLegendaJson ()
 
void postBeginJob ()
 
void postEndJob ()
 
void postEvent (edm::StreamContext const &)
 
void postGlobalBeginRun (edm::GlobalContext const &)
 
void postGlobalEndLumi (edm::GlobalContext const &)
 
void postModuleEvent (edm::StreamContext const &, edm::ModuleCallingContext const &)
 
void postSourceEvent (edm::StreamID)
 
void postStreamBeginLumi (edm::StreamContext const &)
 
void postStreamEndLumi (edm::StreamContext const &)
 
void preallocate (edm::service::SystemBounds const &)
 
void preEvent (edm::StreamContext const &)
 
void preGlobalBeginLumi (edm::GlobalContext const &)
 
void preGlobalEarlyTermination (edm::GlobalContext const &, edm::TerminationOrigin)
 
void preGlobalEndLumi (edm::GlobalContext const &)
 
void preModuleBeginJob (edm::ModuleDescription const &)
 
void preModuleEvent (edm::StreamContext const &, edm::ModuleCallingContext const &)
 
void prePathEvent (edm::StreamContext const &, edm::PathContext const &)
 
void preSourceEarlyTermination (edm::TerminationOrigin)
 
void preSourceEvent (edm::StreamID)
 
void preStreamBeginLumi (edm::StreamContext const &)
 
void preStreamEarlyTermination (edm::StreamContext const &, edm::TerminationOrigin)
 
void preStreamEndLumi (edm::StreamContext const &)
 
void reportEventsThisLumiInSource (unsigned int lumi, unsigned int events)
 
void reportLockWait (unsigned int ls, double waitTime, unsigned int lockCount)
 
void setExceptionDetected (unsigned int ls)
 
void setMicroState (MicroStateService::Microstate)
 
void setMicroState (edm::StreamID, MicroStateService::Microstate)
 
bool shouldWriteFiles (unsigned int lumi, unsigned int *proc=0)
 
void startedLookingForFile ()
 
void stoppedLookingForFile (unsigned int lumi)
 
 ~FastMonitoringService ()
 
- Public Member Functions inherited from evf::MicroStateService
virtual std::string getMicroState1 ()
 
virtual std::string const & getMicroState2 ()
 
 MicroStateService (const edm::ParameterSet &, edm::ActivityRegistry &)
 
virtual ~MicroStateService ()
 

Static Public Member Functions

static void fillDescriptions (edm::ConfigurationDescriptions &descriptions)
 

Static Public Attributes

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

Private Member Functions

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

Private Attributes

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

Additional Inherited Members

- Public Types inherited from evf::MicroStateService
enum  Microstate {
  mInvalid = 0, mFwkOvh, mIdle, mInput,
  mInputDone, mDqm, mEoL, mCOUNT
}
 
- Static Protected Attributes inherited from evf::MicroStateService
static const std::string default_return_ ="NotImplemented"
 

Detailed Description

Definition at line 55 of file FastMonitoringService.h.

Constructor & Destructor Documentation

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

Definition at line 39 of file FastMonitoringService.cc.

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

40  :
41  MicroStateService(iPS,reg)
43  ,nStreams_(0)//until initialized
44  ,sleepTime_(iPS.getUntrackedParameter<int>("sleepTime", 1))
45  ,fastMonIntervals_(iPS.getUntrackedParameter<unsigned int>("fastMonIntervals", 2))
46  ,fastName_("fastmoni")
47  ,slowName_("slowmoni")
49  {
50  reg.watchPreallocate(this, &FastMonitoringService::preallocate);//receiving information on number of threads
52 
56 
60 
65 
67 
70 
71  reg.watchPreSourceEvent(this,&FastMonitoringService::preSourceEvent);//source (with streamID of requestor)
73 
76 
80 
81  //find microstate definition path (required by the module)
82  struct stat statbuf;
83  std::string microstateBaseSuffix = "src/EventFilter/Utilities/plugins/microstatedef.jsd";
84  std::string microstatePath = std::string(getenv("CMSSW_BASE")) + "/" + microstateBaseSuffix;
85  if (stat(microstatePath.c_str(), &statbuf)) {
86  microstatePath = std::string(getenv("CMSSW_RELEASE_BASE")) + "/" + microstateBaseSuffix;
87  if (stat(microstatePath.c_str(), &statbuf)) {
88  microstatePath = microstateBaseSuffix;
89  if (stat(microstatePath.c_str(), &statbuf))
90  throw cms::Exception("FastMonitoringService") << "microstate definition file not found";
91  }
92  }
93  fastMicrostateDefPath_ = microstateDefPath_ = microstatePath;
94  }
void prePathEvent(edm::StreamContext const &, edm::PathContext const &)
void watchPreStreamEarlyTermination(PreStreamEarlyTermination::slot_type const &iSlot)
T getUntrackedParameter(std::string const &, T const &) const
void watchPreEvent(PreEvent::slot_type const &iSlot)
void watchPrePathEvent(PrePathEvent::slot_type const &iSlot)
void watchPreallocate(Preallocate::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
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)
#define NRESERVEDMODULES
void watchPostGlobalEndLumi(PostGlobalEndLumi::slot_type const &iSlot)
void watchPreModuleBeginJob(PreModuleBeginJob::slot_type const &iSlot)
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 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
evf::FastMonitoringService::~FastMonitoringService ( )

Definition at line 97 of file FastMonitoringService.cc.

98  {
99  }

Member Function Documentation

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

Definition at line 591 of file FastMonitoringService.cc.

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

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

591  {
592  std::lock_guard<std::mutex> lock(fmt_.monlock_);
593 
594  if (accuSize_.find(lumi)==accuSize_.end()) accuSize_[lumi] = fileSize;
595  else accuSize_[lumi] += fileSize;
596 
599  else
601  }
std::map< unsigned int, unsigned long > accuSize_
tuple lumi
Definition: fjr2json.py:35
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
void evf::FastMonitoringService::doSnapshot ( const unsigned int  ls,
const bool  isGlobalEOL 
)
private

Definition at line 675 of file FastMonitoringService.cc.

References avgLeadTime_, encModule_, evf::FastMonitoringService::Encoding::encode(), encPath_, evf::FastMonitoringThread::MonitorData::fastAvgLeadTimeJ_, evf::FastMonitoringThread::MonitorData::fastFilesProcessedJ_, evf::FastMonitoringThread::MonitorData::fastLockCountJ_, evf::FastMonitoringThread::MonitorData::fastLockWaitJ_, evf::FastMonitoringThread::MonitorData::fastMacrostateJ_, filesProcessedDuringLumi_, fmt_, i, isGlobalLumiTransition_, evf::FastMonitoringThread::jsonMonitor_, lockStatsDuringLumi_, evf::FastMonitoringThread::m_data, macrostate_, microstate_, evf::FastMonitoringThread::MonitorData::microstateEncoded_, ministate_, evf::FastMonitoringThread::MonitorData::ministateEncoded_, and nStreams_.

Referenced by dowork(), and preGlobalEndLumi().

675  {
676  // update macrostate
678 
679  //update these unless in the midst of a global transition
681 
682  auto itd = avgLeadTime_.find(ls);
683  if (itd != avgLeadTime_.end())
684  fmt_.m_data.fastAvgLeadTimeJ_ = itd->second;
685  else fmt_.m_data.fastAvgLeadTimeJ_=0.;
686 
687  auto iti = filesProcessedDuringLumi_.find(ls);
688  if (iti != filesProcessedDuringLumi_.end())
689  fmt_.m_data.fastFilesProcessedJ_ = iti->second;
691 
692  auto itrd = lockStatsDuringLumi_.find(ls);
693  if (itrd != lockStatsDuringLumi_.end()) {
694  fmt_.m_data.fastLockWaitJ_ = itrd->second.first;
695  fmt_.m_data.fastLockCountJ_ = itrd->second.second;
696  }
697  else {
700  }
701 
702  }
703  else return;
704 
705  //capture latest mini/microstate of streams
706  for (unsigned int i=0;i<nStreams_;i++) {
709  }
710  //for (unsigned int i=0;i<nThreads_;i++)
711  // fmt_.m_data.threadMicrostateEncoded_[i] = encModule_.encode(threadMicrostate_[i]);
712 
713  if (isGlobalEOL)
714  {//only update global variables
715  fmt_.jsonMonitor_->snapGlobal(ls);
716  }
717  else
718  fmt_.jsonMonitor_->snap(ls);
719  }
int i
Definition: DBlmapReader.cc:9
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::map< unsigned int, double > avgLeadTime_
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
std::vector< const void * > microstate_
std::vector< unsigned int > microstateEncoded_
FastMonitoringThread::Macrostate macrostate_
std::unique_ptr< FastMonitor > jsonMonitor_
std::vector< const void * > ministate_
std::vector< Encoding > encPath_
std::vector< unsigned int > ministateEncoded_
void evf::FastMonitoringService::doStreamEOLSnapshot ( const unsigned int  ls,
const unsigned int  streamID 
)
inlineprivate

Definition at line 173 of file FastMonitoringService.h.

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

Referenced by preStreamEndLumi().

173  {
174  //pick up only event count here
175  fmt_.jsonMonitor_->snapStreamAtomic(ls,streamID);
176  }
std::unique_ptr< FastMonitor > jsonMonitor_
void evf::FastMonitoringService::dowork ( )
inlineprivate

Definition at line 178 of file FastMonitoringService.h.

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

Referenced by preallocate().

178  { // the function to be called in the thread. Thread completes when function returns.
179  monInit_.exchange(true,std::memory_order_acquire);
180  while (!fmt_.m_stoprequest) {
181  edm::LogInfo("FastMonitoringService") << "Current states: Ms=" << fmt_.m_data.fastMacrostateJ_.value()
182  << " ms=" << encPath_[0].encode(ministate_[0])
183  << " us=" << encModule_.encode(microstate_[0]) << std::endl;
184 
185  {
186  std::lock_guard<std::mutex> lock(fmt_.monlock_);
187 
189 
191  std::string CSV = fmt_.jsonMonitor_->getCSVString();
192  //release mutex before writing out fast path file
193  fmt_.monlock_.unlock();
194  if (CSV.size())
195  fmt_.jsonMonitor_->outputCSV(fastPath_,CSV);
196  }
197 
198  snapCounter_++;
199 
200  }
201  ::sleep(sleepTime_);
202  }
203  }
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
std::vector< const void * > microstate_
std::atomic< bool > m_stoprequest
std::unique_ptr< FastMonitor > jsonMonitor_
std::vector< const void * > ministate_
std::vector< Encoding > encPath_
void evf::FastMonitoringService::fillDescriptions ( edm::ConfigurationDescriptions descriptions)
static

Definition at line 101 of file FastMonitoringService.cc.

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

102  {
104  desc.setComment("Service for File-based DAQ monitoring and event accounting");
105  desc.addUntracked<int> ("sleepTime",1)->setComment("Sleep time of the monitoring thread");
106  desc.addUntracked<unsigned int> ("fastMonIntervals",2)->setComment("Modulo of sleepTime intervals on which fastmon file is written out");
107  desc.setAllowAnything();
108  descriptions.add("FastMonitoringService", desc);
109  }
ParameterDescriptionBase * addUntracked(U const &iLabel, T const &value)
void setAllowAnything()
allow any parameter label/value pairs
void setComment(std::string const &value)
void add(std::string const &label, ParameterSetDescription const &psetDescription)
bool evf::FastMonitoringService::getAbortFlagForLumi ( unsigned int  lumi)

Definition at line 661 of file FastMonitoringService.cc.

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

Referenced by shouldWriteFiles().

661  {
662  std::lock_guard<std::mutex> lock(fmt_.monlock_);
663 
664  auto it = processedEventsPerLumi_.find(lumi);
665  if (it!=processedEventsPerLumi_.end()) {
666  unsigned int abortFlag = it->second.second;
667  return abortFlag;
668  }
669  else {
670  throw cms::Exception("FastMonitoringService") << "output module wants already deleted (or never reported by SOURCE) lumisection status for LUMI -: "<<lumi;
671  return 0;
672  }
673  }
tuple lumi
Definition: fjr2json.py:35
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
unsigned int evf::FastMonitoringService::getEventsProcessedForLumi ( unsigned int  lumi,
bool *  abortFlag = 0 
)

Definition at line 645 of file FastMonitoringService.cc.

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

Referenced by dqm::DQMFileSaverPB::fillJson(), DQMFileSaver::fillJson(), DQMFileSaver::saveForFilterUnit(), dqm::DQMFileSaverPB::saveLumi(), and shouldWriteFiles().

645  {
646  std::lock_guard<std::mutex> lock(fmt_.monlock_);
647 
648  auto it = processedEventsPerLumi_.find(lumi);
649  if (it!=processedEventsPerLumi_.end()) {
650  unsigned int proc = it->second.first;
651  if (abortFlag) *abortFlag=it->second.second;
652  return proc;
653  }
654  else {
655  throw cms::Exception("FastMonitoringService") << "output module wants already deleted (or never reported by SOURCE) lumisection event count for LUMI -: "<<lumi;
656  return 0;
657  }
658  }
TrainProcessor *const proc
Definition: MVATrainer.cc:101
tuple lumi
Definition: fjr2json.py:35
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
std::string evf::FastMonitoringService::getRunDirName ( ) const
inline

Definition at line 167 of file FastMonitoringService.h.

References runDirectory_.

167 { return runDirectory_.stem().string(); }
boost::filesystem::path runDirectory_
void evf::FastMonitoringService::jobFailure ( )
std::string evf::FastMonitoringService::makeModuleLegendaJson ( )

Definition at line 124 of file FastMonitoringService.cc.

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

Referenced by postBeginJob().

124  {
125  Json::Value legendaVector(Json::arrayValue);
126  for(int i = 0; i < encModule_.current_; i++)
127  legendaVector.append(Json::Value(((const edm::ModuleDescription *)(encModule_.decode(i)))->moduleLabel()));
128  Json::Value valReserved(NRESERVEDMODULES);
129  Json::Value valSpecial(NSPECIALMODULES);
130  Json::Value valOutputModules(nOutputModules_);
131  Json::Value moduleLegend;
132  moduleLegend["names"]=legendaVector;
133  moduleLegend["reserved"]=valReserved;
134  moduleLegend["special"]=valSpecial;
135  moduleLegend["output"]=valOutputModules;
136  Json::StyledWriter writer;
137  return writer.write(moduleLegend);
138  }
int i
Definition: DBlmapReader.cc:9
Represents a JSON value.
Definition: value.h:111
#define NRESERVEDMODULES
virtual std::string write(const Value &root)
Serialize a Value in JSON format.
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:65
const void * decode(unsigned int index)
#define NSPECIALMODULES
array value (ordered list)
Definition: value.h:31
std::string evf::FastMonitoringService::makePathLegendaJson ( )

Definition at line 112 of file FastMonitoringService.cc.

References Json::Value::append(), Json::arrayValue, encPath_, i, NRESERVEDPATHS, AlCaHLTBitMon_QueryRunRegistry::string, and Json::StyledWriter::write().

Referenced by prePathEvent().

112  {
113  Json::Value legendaVector(Json::arrayValue);
114  for(int i = 0; i < encPath_[0].current_; i++)
115  legendaVector.append(Json::Value(*((std::string *)(encPath_[0].decode(i)))));
116  Json::Value valReserved(NRESERVEDPATHS);
117  Json::Value pathLegend;
118  pathLegend["names"]=legendaVector;
119  pathLegend["reserved"]=valReserved;
120  Json::StyledWriter writer;
121  return writer.write(pathLegend);
122  }
int i
Definition: DBlmapReader.cc:9
Represents a JSON value.
Definition: value.h:111
virtual std::string write(const Value &root)
Serialize a Value in JSON format.
Writes a Value in JSON format in a human friendly way.
Definition: writer.h:65
#define NRESERVEDPATHS
std::vector< Encoding > encPath_
array value (ordered list)
Definition: value.h:31
void evf::FastMonitoringService::postBeginJob ( )

Definition at line 320 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

321  {
322  std::string && moduleLegStrJson = makeModuleLegendaJson();
323  FileIO::writeStringToFile(moduleLegendFileJson_, moduleLegStrJson);
324 
326 
327  //update number of entries in module histogram
328  std::lock_guard<std::mutex> lock(fmt_.monlock_);
330  }
FastMonitoringThread::Macrostate macrostate_
void evf::FastMonitoringService::postEndJob ( )
void evf::FastMonitoringService::postEvent ( edm::StreamContext const &  sc)

Definition at line 526 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

527  {
528  microstate_[sc.streamID()] = &reservedMicroStateNames[mIdle];
529 
530  ministate_[sc.streamID()] = &nopath_;
531 
532  #if ATOMIC_LEVEL>=2
533  //use atomic flag to make sure end of lumi sees this
534  streamCounterUpdating_[sc.streamID()]->store(true,std::memory_order_release);
535  fmt_.m_data.processed_[sc.streamID()]->fetch_add(1,std::memory_order_release);
536  streamCounterUpdating_[sc.streamID()]->store(false,std::memory_order_release);
537 
538  #elif ATOMIC_LEVEL==1
539  //writes are atomic, we assume writes propagate to memory before stream EOL snap
540  fmt_.m_data.processed_[sc.streamID()]->fetch_add(1,std::memory_order_relaxed);
541 
542  #elif ATOMIC_LEVEL==0 //default
543  (*(fmt_.m_data.processed_[sc.streamID()]))++;
544  #endif
545  eventCountForPathInit_[sc.streamID()]++;
546 
547  //fast path counter (events accumulated in a run)
548  unsigned long res = totalEventsProcessed_.fetch_add(1,std::memory_order_relaxed);
550  //fmt_.m_data.fastPathProcessedJ_ = totalEventsProcessed_.load(std::memory_order_relaxed);
551  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< AtomicMonUInt * > processed_
std::vector< unsigned int > eventCountForPathInit_
std::vector< std::atomic< bool > * > streamCounterUpdating_
static const std::string nopath_
std::vector< const void * > microstate_
std::atomic< unsigned long > totalEventsProcessed_
std::vector< const void * > ministate_
void evf::FastMonitoringService::postGlobalBeginRun ( edm::GlobalContext const &  gc)
void evf::FastMonitoringService::postGlobalEndLumi ( edm::GlobalContext const &  gc)

Definition at line 443 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

444  {
445  //mark closed lumis (still keep map entries until next one)
446  lastGlobalLumisClosed_.push(gc.luminosityBlockID().luminosityBlock());
447  }
std::queue< unsigned int > lastGlobalLumisClosed_
void evf::FastMonitoringService::postModuleEvent ( edm::StreamContext const &  sc,
edm::ModuleCallingContext const &  mcc 
)

Definition at line 568 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

569  {
570  //microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
571  microstate_[sc.streamID().value()] = &reservedMicroStateNames[mFwkOvh];
572  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< const void * > microstate_
void evf::FastMonitoringService::postSourceEvent ( edm::StreamID  sid)

Definition at line 558 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

559  {
561  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< const void * > microstate_
unsigned int value() const
Definition: StreamID.h:46
void evf::FastMonitoringService::postStreamBeginLumi ( edm::StreamContext const &  sc)
void evf::FastMonitoringService::postStreamEndLumi ( edm::StreamContext const &  sc)
void evf::FastMonitoringService::preallocate ( edm::service::SystemBounds const &  bounds)

Definition at line 140 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

141  {
142 
143  // FIND RUN DIRECTORY
144  // The run dir should be set via the configuration of EvFDaqDirector
145 
146  if (edm::Service<evf::EvFDaqDirector>().operator->()==nullptr)
147  {
148  throw cms::Exception("FastMonitoringService") << "EvFDaqDirector is not present";
149 
150  }
151  emptyLumisectionMode_ = edm::Service<evf::EvFDaqDirector>()->emptyLumisectionMode();
152  boost::filesystem::path runDirectory(edm::Service<evf::EvFDaqDirector>()->baseRunDir());
153  workingDirectory_ = runDirectory_ = runDirectory;
154  workingDirectory_ /= "mon";
155 
156  if ( !boost::filesystem::is_directory(workingDirectory_)) {
157  LogDebug("FastMonitoringService") << "<MON> DIR NOT FOUND! Trying to create -: " << workingDirectory_.string() ;
158  boost::filesystem::create_directories(workingDirectory_);
159  if ( !boost::filesystem::is_directory(workingDirectory_))
160  edm::LogWarning("FastMonitoringService") << "Unable to create <MON> DIR -: " << workingDirectory_.string()
161  << ". No monitoring data will be written.";
162  }
163 
164  std::ostringstream fastFileName;
165 
166  fastFileName << fastName_ << "_pid" << std::setfill('0') << std::setw(5) << getpid() << ".fast";
168  fast /= fastFileName.str();
169  fastPath_ = fast.string();
170 
171  std::ostringstream moduleLegFile;
172  std::ostringstream moduleLegFileJson;
173  moduleLegFile << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
174  moduleLegFileJson << "microstatelegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
175  moduleLegendFile_ = (workingDirectory_/moduleLegFile.str()).string();
176  moduleLegendFileJson_ = (workingDirectory_/moduleLegFileJson.str()).string();
177 
178  std::ostringstream pathLegFile;
179  std::ostringstream pathLegFileJson;
180  pathLegFile << "pathlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".leg";
181  pathLegendFile_ = (workingDirectory_/pathLegFile.str()).string();
182  pathLegFileJson << "pathlegend_pid" << std::setfill('0') << std::setw(5) << getpid() << ".jsn";
183  pathLegendFileJson_ = (workingDirectory_/pathLegFileJson.str()).string();
184 
185  LogDebug("FastMonitoringService") << "Initializing FastMonitor with microstate def path -: "
187  //<< encPath_.current_ + 1 << " " << encModule_.current_ + 1
188 
189  nStreams_=bounds.maxNumberOfStreams();
190  nThreads_=bounds.maxNumberOfThreads();
191 
192  //this should already be >=1
193  if (nStreams_==0) nStreams_=1;
194  if (nThreads_==0) nThreads_=1;
195 
196  /*
197  * initialize the fast monitor with:
198  * vector of pointers to monitorable parameters
199  * path to definition
200  *
201  */
202 
204 
205  for(unsigned int i = 0; i < (mCOUNT); i++)
208 
209  for (unsigned int i=0;i<nStreams_;i++) {
210  ministate_.push_back(&nopath_);
212 
213  //for synchronization
214  streamCounterUpdating_.push_back(new std::atomic<bool>(0));
215 
216  //path (mini) state
217  encPath_.emplace_back(0);
218  encPath_[i].update((void*)&nopath_);
219  eventCountForPathInit_.push_back(0);
220  firstEventId_.push_back(0);
221  collectedPathList_.push_back(new std::atomic<bool>(0));
222 
223  }
224  //for (unsigned int i=0;i<nThreads_;i++)
225  // threadMicrostate_.push_back(&reservedMicroStateNames[mInvalid]);
226 
227  //initial size until we detect number of bins
231 
232  lastGlobalLumi_=0;
234  lumiFromSource_=0;
235 
236  //startup monitoring
238  fmt_.jsonMonitor_->setNStreams(nStreams_);
240  monInit_.store(false,std::memory_order_release);
242 
243  //this definition needs: #include "tbb/compat/thread"
244  //however this would results in TBB imeplementation replacing std::thread
245  //(both supposedly call pthread_self())
246  //number of threads created in process could be obtained from /proc,
247  //assuming that all posix threads are true kernel threads capable of running in parallel
248 
249  //#if TBB_IMPLEMENT_CPP0X
251  //threadIDAvailable_=true;
252  //#endif
253 
254  }
#define LogDebug(id)
int i
Definition: DBlmapReader.cc:9
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
boost::filesystem::path runDirectory_
void start(void(FastMonitoringService::*fp)(), FastMonitoringService *cp)
std::vector< unsigned int > eventCountForPathInit_
std::vector< std::atomic< bool > * > streamCounterUpdating_
std::vector< std::atomic< bool > * > collectedPathList_
tuple path
else: Piece not in the list, fine.
static const std::string nopath_
void resetFastMonitor(std::string const &microStateDefPath, std::string const &fastMicroStateDefPath)
std::vector< unsigned long > firstEventId_
std::vector< const void * > microstate_
void registerVariables(FastMonitor *fm, unsigned int nStreams, unsigned int nThreads)
FastMonitoringThread::Macrostate macrostate_
boost::filesystem::path workingDirectory_
std::unique_ptr< FastMonitor > jsonMonitor_
std::vector< const void * > ministate_
std::vector< Encoding > encPath_
void evf::FastMonitoringService::preEvent ( edm::StreamContext const &  sc)

Definition at line 522 of file FastMonitoringService.cc.

Referenced by FastMonitoringService().

523  {
524  }
void evf::FastMonitoringService::preGlobalBeginLumi ( edm::GlobalContext const &  gc)

Definition at line 343 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

344  {
345 
346  timeval lumiStartTime;
347  gettimeofday(&lumiStartTime, 0);
348  unsigned int newLumi = gc.luminosityBlockID().luminosityBlock();
349 
350  std::lock_guard<std::mutex> lock(fmt_.monlock_);
351 
352  lumiStartTime_[newLumi]=lumiStartTime;
353  while (!lastGlobalLumisClosed_.empty()) {
354  //wipe out old map entries as they aren't needed and slow down access
355  unsigned int oldLumi = lastGlobalLumisClosed_.back();
357  lumiStartTime_.erase(oldLumi);
358  avgLeadTime_.erase(oldLumi);
359  filesProcessedDuringLumi_.erase(oldLumi);
360  accuSize_.erase(oldLumi);
361  lockStatsDuringLumi_.erase(oldLumi);
362  processedEventsPerLumi_.erase(oldLumi);
363  }
364  lastGlobalLumi_= newLumi;
366  }
std::map< unsigned int, timeval > lumiStartTime_
std::map< unsigned int, unsigned long > accuSize_
std::map< unsigned int, unsigned int > filesProcessedDuringLumi_
std::queue< unsigned int > lastGlobalLumisClosed_
std::map< unsigned int, double > avgLeadTime_
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
void evf::FastMonitoringService::preGlobalEarlyTermination ( edm::GlobalContext const &  gc,
edm::TerminationOrigin  to 
)

Definition at line 269 of file FastMonitoringService.cc.

References edm::ExceptionFromAnotherContext, edm::ExceptionFromThisContext, exceptionInLS_, edm::ExternalSignal, fmt_, CommonMethods::lock(), edm::LuminosityBlockID::luminosityBlock(), edm::GlobalContext::luminosityBlockID(), evf::FastMonitoringThread::monlock_, and AlCaHLTBitMon_QueryRunRegistry::string.

Referenced by FastMonitoringService().

270  {
271  std::string context;
272  if (to==edm::TerminationOrigin::ExceptionFromThisContext) context = " FromThisContext ";
273  if (to==edm::TerminationOrigin::ExceptionFromAnotherContext) context = " FromAnotherContext";
274  if (to==edm::TerminationOrigin::ExternalSignal) context = " FromExternalSignal";
275  edm::LogWarning("FastMonitoringService") << " GLOBAL " << "earlyTermination -: LS:"
276  << gc.luminosityBlockID().luminosityBlock() << " " << context;
277  std::lock_guard<std::mutex> lock(fmt_.monlock_);
278  exceptionInLS_.push_back(gc.luminosityBlockID().luminosityBlock());
279  //exception_detected_=true;
280  }
std::vector< unsigned int > exceptionInLS_
void evf::FastMonitoringService::preGlobalEndLumi ( edm::GlobalContext const &  gc)

Definition at line 368 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

369  {
370  unsigned int lumi = gc.luminosityBlockID().luminosityBlock();
371  LogDebug("FastMonitoringService") << "Lumi ended. Writing JSON information. LUMI -: "
372  << lumi;
373  timeval lumiStopTime;
374  gettimeofday(&lumiStopTime, 0);
375 
376  std::lock_guard<std::mutex> lock(fmt_.monlock_);
377 
378  // Compute throughput
379  timeval stt = lumiStartTime_[lumi];
380  unsigned long usecondsForLumi = (lumiStopTime.tv_sec - stt.tv_sec)*1000000
381  + (lumiStopTime.tv_usec - stt.tv_usec);
382  unsigned long accuSize = accuSize_.find(lumi)==accuSize_.end() ? 0 : accuSize_[lumi];
383  double throughput = throughputFactor()* double(accuSize) / double(usecondsForLumi);
384  //store to registered variable
385  fmt_.m_data.fastThroughputJ_.value() = throughput;
386 
387  //update
388  doSnapshot(lumi,true);
389 
390  // create file name for slow monitoring file
391  std::stringstream slowFileName;
392  slowFileName << slowName_ << "_ls" << std::setfill('0') << std::setw(4)
393  << lumi << "_pid" << std::setfill('0')
394  << std::setw(5) << getpid() << ".jsn";
396  slow /= slowFileName.str();
397 
398  //retrieve one result we need (todo: sanity check if it's found)
399  IntJ *lumiProcessedJptr = dynamic_cast<IntJ*>(fmt_.jsonMonitor_->getMergedIntJForLumi("Processed",lumi));
400  if (!lumiProcessedJptr)
401  throw cms::Exception("FastMonitoringService") << "Internal error: got null pointer from FastMonitor";
402  processedEventsPerLumi_[lumi] = std::pair<unsigned int,bool>(lumiProcessedJptr->value(),false);
403 
404  //checking if exception has been thrown (in case of Global/Stream early termination, for this LS)
405  bool exception_detected = exception_detected_;
406  for (auto ex : exceptionInLS_)
407  if (lumi == ex) exception_detected=true;
408 
409  if (edm::shutdown_flag || exception_detected) {
410  edm::LogInfo("FastMonitoringService") << "Run interrupted. Skip writing EoL information -: "
411  << processedEventsPerLumi_[lumi].first << " events were processed in LUMI " << lumi;
412  //this will prevent output modules from producing json file for possibly incomplete lumi
413  processedEventsPerLumi_[lumi].first=0;
414  processedEventsPerLumi_[lumi].second=true;
415  //disable this exception, so service can be used standalone (will be thrown if output module asks for this information)
416  //throw cms::Exception("FastMonitoringService") << "SOURCE did not send update for lumi block. LUMI -:" << lumi;
417  return;
418 
419  }
420 
421  auto itr = sourceEventsReport_.find(lumi);
422  if (itr!=sourceEventsReport_.end()) {
423  if (itr->second!=processedEventsPerLumi_[lumi].first) {
424  throw cms::Exception("FastMonitoringService") << "MISMATCH with SOURCE update. LUMI -: "
425  << lumi
426  << ", events(processed):" << processedEventsPerLumi_[lumi].first
427  << " events(source):" << itr->second;
428  }
429  sourceEventsReport_.erase(itr);
430  }
431  edm::LogInfo("FastMonitoringService") << "Statistics for lumisection -: lumi = " << lumi << " events = "
432  << lumiProcessedJptr->value() << " time = " << usecondsForLumi/1000000
433  << " size = " << accuSize << " thr = " << throughput;
434  delete lumiProcessedJptr;
435 
436  //full global and stream merge&output for this lumi
437  fmt_.jsonMonitor_->outputFullJSON(slow.string(),lumi);//full global and stream merge and JSON write for this lumi
438  fmt_.jsonMonitor_->discardCollected(lumi);//we don't do further updates for this lumi
439 
441  }
#define LogDebug(id)
std::map< unsigned int, timeval > lumiStartTime_
double throughputFactor()
std::map< unsigned int, unsigned long > accuSize_
tuple lumi
Definition: fjr2json.py:35
volatile std::atomic< bool > shutdown_flag
tuple path
else: Piece not in the list, fine.
void doSnapshot(const unsigned int ls, const bool isGlobalEOL)
std::map< unsigned int, std::pair< unsigned int, bool > > processedEventsPerLumi_
boost::filesystem::path workingDirectory_
std::map< unsigned int, unsigned int > sourceEventsReport_
std::unique_ptr< FastMonitor > jsonMonitor_
std::vector< unsigned int > exceptionInLS_
void evf::FastMonitoringService::preModuleBeginJob ( edm::ModuleDescription const &  desc)

Definition at line 304 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

305  {
306  std::lock_guard<std::mutex> lock(fmt_.monlock_);
307  //std::cout << " Pre module Begin Job module: " << desc.moduleName() << std::endl;
308 
309  //build a map of modules keyed by their module description address
310  //here we need to treat output modules in a special way so they can be easily singled out
311  if(desc.moduleName() == "Stream" || desc.moduleName() == "ShmStreamConsumer" || desc.moduleName() == "EvFOutputModule" ||
312  desc.moduleName() == "EventStreamFileWriter" || desc.moduleName() == "PoolOutputModule") {
313  encModule_.updateReserved((void*)&desc);
314  nOutputModules_++;
315  }
316  else
317  encModule_.update((void*)&desc);
318  }
void evf::FastMonitoringService::preModuleEvent ( edm::StreamContext const &  sc,
edm::ModuleCallingContext const &  mcc 
)

Definition at line 563 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

564  {
565  microstate_[sc.streamID().value()] = (void*)(mcc.moduleDescription());
566  }
std::vector< const void * > microstate_
void evf::FastMonitoringService::prePathEvent ( edm::StreamContext const &  sc,
edm::PathContext const &  pc 
)

Definition at line 489 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

490  {
491  //make sure that all path names are retrieved before allowing ministate to change
492  //hack: assume memory is synchronized after ~50 events seen by each stream
493  if (unlikely(eventCountForPathInit_[sc.streamID()]<50) && false==collectedPathList_[sc.streamID()]->load(std::memory_order_acquire))
494  {
495  //protection between stream threads, as well as the service monitoring thread
496  std::lock_guard<std::mutex> lock(fmt_.monlock_);
497 
498  if (firstEventId_[sc.streamID()]==0)
499  firstEventId_[sc.streamID()]=sc.eventID().event();
500  if (sc.eventID().event()==firstEventId_[sc.streamID()])
501  {
502  encPath_[sc.streamID()].update((void*)&pc.pathName());
503  return;
504  }
505  else {
506  //finished collecting path names
507  collectedPathList_[sc.streamID()]->store(true,std::memory_order_seq_cst);
508  fmt_.m_data.ministateBins_=encPath_[sc.streamID()].vecsize();
509  if (!pathLegendWritten_) {
510  std::string pathLegendStrJson = makePathLegendaJson();
511  FileIO::writeStringToFile(pathLegendFileJson_, pathLegendStrJson);
512  pathLegendWritten_=true;
513  }
514  }
515  }
516  else {
517  ministate_[sc.streamID()] = &(pc.pathName());
518  }
519  }
std::vector< unsigned int > eventCountForPathInit_
#define unlikely(x)
std::vector< std::atomic< bool > * > collectedPathList_
std::vector< unsigned long > firstEventId_
std::vector< const void * > ministate_
std::vector< Encoding > encPath_
void evf::FastMonitoringService::preSourceEarlyTermination ( edm::TerminationOrigin  to)
void evf::FastMonitoringService::preSourceEvent ( edm::StreamID  sid)

Definition at line 553 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

554  {
556  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< const void * > microstate_
unsigned int value() const
Definition: StreamID.h:46
void evf::FastMonitoringService::preStreamBeginLumi ( edm::StreamContext const &  sc)

Definition at line 449 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

450  {
451  unsigned int sid = sc.streamID().value();
452  std::lock_guard<std::mutex> lock(fmt_.monlock_);
453  fmt_.m_data.streamLumi_[sid] = sc.eventID().luminosityBlock();
454 
455  //reset collected values for this stream
456  *(fmt_.m_data.processed_[sid])=0;
457 
458  ministate_[sid]=&nopath_;
460  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< AtomicMonUInt * > processed_
static const std::string nopath_
std::vector< const void * > microstate_
std::vector< const void * > ministate_
std::vector< unsigned int > streamLumi_
void evf::FastMonitoringService::preStreamEarlyTermination ( edm::StreamContext const &  sc,
edm::TerminationOrigin  to 
)

Definition at line 256 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

257  {
258  std::string context;
259  if (to==edm::TerminationOrigin::ExceptionFromThisContext) context = " FromThisContext ";
260  if (to==edm::TerminationOrigin::ExceptionFromAnotherContext) context = " FromAnotherContext";
261  if (to==edm::TerminationOrigin::ExternalSignal) context = " FromExternalSignal";
262  edm::LogWarning("FastMonitoringService") << " STREAM " << sc.streamID().value() << " earlyTermination -: ID:"<< sc.eventID()
263  << " LS:" << sc.eventID().luminosityBlock() << " " << context;
264  std::lock_guard<std::mutex> lock(fmt_.monlock_);
265  exceptionInLS_.push_back(sc.eventID().luminosityBlock());
266  //exception_detected_=true;
267  }
std::vector< unsigned int > exceptionInLS_
void evf::FastMonitoringService::preStreamEndLumi ( edm::StreamContext const &  sc)

Definition at line 467 of file FastMonitoringService.cc.

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

Referenced by FastMonitoringService().

468  {
469  unsigned int sid = sc.streamID().value();
470  std::lock_guard<std::mutex> lock(fmt_.monlock_);
471 
472  #if ATOMIC_LEVEL>=2
473  //spinlock to make sure we are not still updating event counter somewhere
474  while (streamCounterUpdating_[sid]->load(std::memory_order_acquire)) {}
475  #endif
476 
477  //update processed count to be complete at this time
478  doStreamEOLSnapshot(sc.eventID().luminosityBlock(),sid);
479  //reset this in case stream does not get notified of next lumi (we keep processed events only)
480  ministate_[sid]=&nopath_;
482  }
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< std::atomic< bool > * > streamCounterUpdating_
void doStreamEOLSnapshot(const unsigned int ls, const unsigned int streamID)
static const std::string nopath_
def load
Definition: svgfig.py:546
std::vector< const void * > microstate_
std::vector< const void * > ministate_
void evf::FastMonitoringService::reportEventsThisLumiInSource ( unsigned int  lumi,
unsigned int  events 
)

Definition at line 721 of file FastMonitoringService.cc.

References fmt_, CommonMethods::lock(), evf::FastMonitoringThread::monlock_, and sourceEventsReport_.

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

722  {
723 
724  std::lock_guard<std::mutex> lock(fmt_.monlock_);
725  auto itr = sourceEventsReport_.find(lumi);
726  if (itr!=sourceEventsReport_.end())
727  itr->second+=events;
728  else
730 
731  }
tuple lumi
Definition: fjr2json.py:35
tuple events
Definition: patZpeak.py:19
std::map< unsigned int, unsigned int > sourceEventsReport_
void evf::FastMonitoringService::reportLockWait ( unsigned int  ls,
double  waitTime,
unsigned int  lockCount 
)

Definition at line 637 of file FastMonitoringService.cc.

References fmt_, CommonMethods::lock(), lockStatsDuringLumi_, python.rootplot.utilities::ls(), and evf::FastMonitoringThread::monlock_.

Referenced by FedRawDataInputSource::readSupervisor().

638  {
639  std::lock_guard<std::mutex> lock(fmt_.monlock_);
640  lockStatsDuringLumi_[ls]=std::pair<double,unsigned int>(waitTime,lockCount);
641 
642  }
std::map< unsigned int, std::pair< double, unsigned int > > lockStatsDuringLumi_
void evf::FastMonitoringService::setExceptionDetected ( unsigned int  ls)

Definition at line 293 of file FastMonitoringService.cc.

References exception_detected_, and exceptionInLS_.

Referenced by FedRawDataInputSource::getNextEvent().

293  {
294  if (!ls) exception_detected_=true;
295  else exceptionInLS_.push_back(ls);
296  }
std::vector< unsigned int > exceptionInLS_
void evf::FastMonitoringService::setMicroState ( MicroStateService::Microstate  m)
virtual

Implements evf::MicroStateService.

Definition at line 578 of file FastMonitoringService.cc.

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

579  {
580  for (unsigned int i=0;i<nStreams_;i++)
582  }
int i
Definition: DBlmapReader.cc:9
static const edm::ModuleDescription reservedMicroStateNames[mCOUNT]
std::vector< const void * > microstate_
void evf::FastMonitoringService::setMicroState ( edm::StreamID  sid,
MicroStateService::Microstate  m 
)
virtual
bool evf::FastMonitoringService::shouldWriteFiles ( unsigned int  lumi,
unsigned int *  proc = 0 
)
inline

Definition at line 161 of file FastMonitoringService.h.

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

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

162  {
163  unsigned int processed = getEventsProcessedForLumi(lumi);
164  if (proc) *proc = processed;
165  return !getAbortFlagForLumi(lumi) && (processed || emptyLumisectionMode_);
166  }
unsigned int getEventsProcessedForLumi(unsigned int lumi, bool *abortFlag=0)
TrainProcessor *const proc
Definition: MVATrainer.cc:101
tuple lumi
Definition: fjr2json.py:35
bool getAbortFlagForLumi(unsigned int lumi)
void evf::FastMonitoringService::startedLookingForFile ( )

Definition at line 603 of file FastMonitoringService.cc.

References fileLookStart_.

Referenced by FedRawDataInputSource::readSupervisor().

603  {
604  gettimeofday(&fileLookStart_, 0);
605  /*
606  std::cout << "Started looking for .raw file at: s=" << fileLookStart_.tv_sec << ": ms = "
607  << fileLookStart_.tv_usec / 1000.0 << std::endl;
608  */
609  }
void evf::FastMonitoringService::stoppedLookingForFile ( unsigned int  lumi)

Definition at line 611 of file FastMonitoringService.cc.

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

Referenced by FedRawDataInputSource::readSupervisor().

611  {
612  gettimeofday(&fileLookStop_, 0);
613  /*
614  std::cout << "Stopped looking for .raw file at: s=" << fileLookStop_.tv_sec << ": ms = "
615  << fileLookStop_.tv_usec / 1000.0 << std::endl;
616  */
617  std::lock_guard<std::mutex> lock(fmt_.monlock_);
618 
619  if (lumi>lumiFromSource_) {
621  leadTimes_.clear();
622  }
623  unsigned long elapsedTime = (fileLookStop_.tv_sec - fileLookStart_.tv_sec) * 1000000 // sec to us
624  + (fileLookStop_.tv_usec - fileLookStart_.tv_usec); // us
625  // add this to lead times for this lumi
626  leadTimes_.push_back((double)elapsedTime);
627 
628  // recompute average lead time for this lumi
629  if (leadTimes_.size() == 1) avgLeadTime_[lumi] = leadTimes_[0];
630  else {
631  double totTime = 0;
632  for (unsigned int i = 0; i < leadTimes_.size(); i++) totTime += leadTimes_[i];
633  avgLeadTime_[lumi] = 0.001*(totTime / leadTimes_.size());
634  }
635  }
int i
Definition: DBlmapReader.cc:9
tuple lumi
Definition: fjr2json.py:35
std::map< unsigned int, double > avgLeadTime_
std::vector< double > leadTimes_

Member Data Documentation

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

Definition at line 238 of file FastMonitoringService.h.

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

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

Definition at line 253 of file FastMonitoringService.h.

Referenced by preallocate(), and prePathEvent().

bool evf::FastMonitoringService::emptyLumisectionMode_ = false
private

Definition at line 275 of file FastMonitoringService.h.

Referenced by preallocate(), and shouldWriteFiles().

Encoding evf::FastMonitoringService::encModule_
private
std::vector<Encoding> evf::FastMonitoringService::encPath_
private
std::vector<unsigned int> evf::FastMonitoringService::eventCountForPathInit_
private

Definition at line 254 of file FastMonitoringService.h.

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

bool evf::FastMonitoringService::exception_detected_ = false
private
std::vector<unsigned int> evf::FastMonitoringService::exceptionInLS_
private
std::string evf::FastMonitoringService::fastMicrostateDefPath_
private

Definition at line 215 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preallocate().

unsigned int evf::FastMonitoringService::fastMonIntervals_
private

Definition at line 213 of file FastMonitoringService.h.

Referenced by dowork().

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

Definition at line 216 of file FastMonitoringService.h.

Referenced by preallocate().

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

Definition at line 216 of file FastMonitoringService.h.

Referenced by dowork(), and preallocate().

timeval evf::FastMonitoringService::fileLookStart_
private

Definition at line 221 of file FastMonitoringService.h.

Referenced by startedLookingForFile(), and stoppedLookingForFile().

timeval evf::FastMonitoringService::fileLookStop_
private

Definition at line 221 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

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

Definition at line 239 of file FastMonitoringService.h.

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

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

Definition at line 252 of file FastMonitoringService.h.

Referenced by preallocate(), and prePathEvent().

FastMonitoringThread evf::FastMonitoringService::fmt_
private
bool evf::FastMonitoringService::isGlobalLumiTransition_
private
unsigned int evf::FastMonitoringService::lastGlobalLumi_
private

Definition at line 223 of file FastMonitoringService.h.

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

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

Definition at line 224 of file FastMonitoringService.h.

Referenced by postGlobalEndLumi(), and preGlobalBeginLumi().

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

Definition at line 242 of file FastMonitoringService.h.

Referenced by stoppedLookingForFile().

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

Definition at line 243 of file FastMonitoringService.h.

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

unsigned int evf::FastMonitoringService::lumiFromSource_
private

Definition at line 226 of file FastMonitoringService.h.

Referenced by preallocate(), and stoppedLookingForFile().

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

Definition at line 220 of file FastMonitoringService.h.

Referenced by preGlobalBeginLumi(), and preGlobalEndLumi().

FastMonitoringThread::Macrostate evf::FastMonitoringService::macrostate_
private
const std::string evf::FastMonitoringService::macroStateNames
static
Initial value:
=
{"Init","JobReady","RunGiven","Running",
"Stopping","Done","JobEnded","Error","ErrorEnded","End",
"Invalid"}

Definition at line 112 of file FastMonitoringService.h.

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

Definition at line 215 of file FastMonitoringService.h.

Referenced by FastMonitoringService(), and preallocate().

std::vector<const void*> evf::FastMonitoringService::ministate_
private
std::string evf::FastMonitoringService::moduleLegendFile_
private

Definition at line 265 of file FastMonitoringService.h.

Referenced by preallocate().

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

Definition at line 266 of file FastMonitoringService.h.

Referenced by postBeginJob(), and preallocate().

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

Definition at line 272 of file FastMonitoringService.h.

Referenced by dowork(), and preallocate().

const std::string evf::FastMonitoringService::nopath_ = "NoPath"
static
unsigned int evf::FastMonitoringService::nOutputModules_ =0
private

Definition at line 270 of file FastMonitoringService.h.

Referenced by makeModuleLegendaJson(), and preModuleBeginJob().

unsigned int evf::FastMonitoringService::nStreams_
private

Definition at line 210 of file FastMonitoringService.h.

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

unsigned int evf::FastMonitoringService::nThreads_
private

Definition at line 211 of file FastMonitoringService.h.

Referenced by preallocate().

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

Definition at line 267 of file FastMonitoringService.h.

Referenced by preallocate().

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

Definition at line 268 of file FastMonitoringService.h.

Referenced by preallocate(), and prePathEvent().

bool evf::FastMonitoringService::pathLegendWritten_ = false
private

Definition at line 269 of file FastMonitoringService.h.

Referenced by prePathEvent().

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

Definition at line 255 of file FastMonitoringService.h.

std::map<unsigned int, std::pair<unsigned int,bool> > evf::FastMonitoringService::processedEventsPerLumi_
private
boost::filesystem::path evf::FastMonitoringService::runDirectory_
private

Definition at line 257 of file FastMonitoringService.h.

Referenced by getRunDirName(), and preallocate().

int evf::FastMonitoringService::sleepTime_
private

Definition at line 212 of file FastMonitoringService.h.

Referenced by dowork().

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

Definition at line 216 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi().

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

Definition at line 214 of file FastMonitoringService.h.

Referenced by dowork().

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

Definition at line 259 of file FastMonitoringService.h.

Referenced by preGlobalEndLumi(), and reportEventsThisLumiInSource().

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

Definition at line 250 of file FastMonitoringService.h.

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

bool evf::FastMonitoringService::threadIDAvailable_ = false
private

Definition at line 261 of file FastMonitoringService.h.

Referenced by preallocate().

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

Definition at line 234 of file FastMonitoringService.h.

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

Definition at line 263 of file FastMonitoringService.h.

Referenced by postEvent().

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

Definition at line 257 of file FastMonitoringService.h.

Referenced by preallocate(), and preGlobalEndLumi().