FastMonitor.cc

Go to the documentation of this file.

/*
 * FastMonitor.cc
 *
 *  Created on: Nov 27, 2012
 *      Author: aspataru
 */

#include "EventFilter/Utilities/interface/FastMonitor.h"
#include "EventFilter/Utilities/interface/JsonSerializable.h"
#include "EventFilter/Utilities/interface/FileIO.h"
#include "FWCore/MessageLogger/interface/MessageLogger.h"

#include <fstream>
#include <iostream>
#include <sstream>
#include <cassert>
#include <sys/types.h>
#include <unistd.h>
#include <boost/filesystem/fstream.hpp>

using namespace jsoncollector;

FastMonitor::FastMonitor(std::string const& defPath, std::string const defGroup, bool strictChecking, bool useSource, bool useDefinition) :
    defPath_(defPath),strictChecking_(strictChecking),useSource_(useSource),useDefinition_(useDefinition),nStreams_(1),deleteDef_(true)
{
  //get host and PID info
  if (useSource)
    getHostAndPID(sourceInfo_);

  //load definition file
  dpd_ = new DataPointDefinition();
  DataPointDefinition::getDataPointDefinitionFor(defPath_, dpd_,&defGroup);

}


FastMonitor::FastMonitor(DataPointDefinition * dpd, bool strictChecking, bool useSource, bool useDefinition) :
    strictChecking_(strictChecking),useSource_(useSource),useDefinition_(useDefinition),nStreams_(1),dpd_(dpd)
{
  //get host and PID info
  if (useSource)
    getHostAndPID(sourceInfo_);
}

FastMonitor::~FastMonitor()
{
  for (auto dp: dataPoints_) delete dp;
  if (deleteDef_) delete dpd_;
  if (deleteDefFast_) delete dpdFast_;
}

void FastMonitor::addFastPathDefinition(std::string const& defPathFast, std::string const defGroupFast, bool strict)
{
  haveFastPath_=true;
  defPathFast_=defPathFast;
  dpdFast_ = new DataPointDefinition();
  DataPointDefinition::getDataPointDefinitionFor(defPathFast_, dpdFast_,&defGroupFast);
  fastPathStrictChecking_=strict;
  deleteDefFast_=true;
}

//per-process variables
void FastMonitor::registerGlobalMonitorable(JsonMonitorable *newMonitorable, bool NAifZeroUpdates, unsigned int *nBins)
{
  DataPoint *dp = new DataPoint(sourceInfo_,defPath_);
  dp->trackMonitorable(newMonitorable,NAifZeroUpdates);
  dp->setNBins(nBins);
  dataPoints_.push_back(dp);
  dpNameMap_[newMonitorable->getName()]=dataPoints_.size()-1;

  //checks if the same name is registered twice
  assert(uids_.insert(newMonitorable->getName()).second);
}

//fast path: no merge operation is performed
void FastMonitor::registerFastGlobalMonitorable(JsonMonitorable *newMonitorable)
{
  DataPoint *dp = new DataPoint(sourceInfo_,defPathFast_,true);
  dp->trackMonitorable(newMonitorable,false);
  dataPointsFastOnly_.push_back(dp);
}

//per-stream variables
void FastMonitor::registerStreamMonitorableUIntVec(std::string const& name, std::vector<unsigned int> *inputs, bool NAifZeroUpdates, unsigned int *nBins)
{
  DataPoint *dp = new DataPoint(sourceInfo_,defPath_);
  dp->trackVectorUInt(name,inputs,NAifZeroUpdates);
  dp->setNBins(nBins);
  dataPoints_.push_back(dp);
  dpNameMap_[name]=dataPoints_.size()-1;
  assert (uids_.insert(name).second);
}


//atomic variables with guaranteed updates at the time of reading
void FastMonitor::registerStreamMonitorableUIntVecAtomic(std::string const& name, std::vector<AtomicMonUInt*> *inputs, bool NAifZeroUpdates, unsigned int *nBins)
{
  std::string definitionToPass;
  if (useDefinition_) definitionToPass=defPath_;
  DataPoint *dp = new DataPoint(definitionToPass,sourceInfo_);
  dp->trackVectorUIntAtomic(name,inputs,NAifZeroUpdates);
  dp->setNBins(nBins);
  dataPoints_.push_back(dp);
  dpNameMap_[name]=dataPoints_.size()-1;
  assert (uids_.insert(name).second);
}



void FastMonitor::commit(std::vector<unsigned int> *streamLumisPtr)
{
  std::vector<std::string> const& jsonNames= dpd_->getNames();
  regDpCount_ = dataPoints_.size();
  if (strictChecking_)
    assert(jsonNames.size()==regDpCount_);

  std::map<unsigned int,bool> hasJson;
  for (unsigned int i=0;i<jsonNames.size();i++)
  {
    bool notFoundVar=true;
    for (unsigned int j=0;j<regDpCount_;j++) {
      if (dataPoints_[j]->getName()==jsonNames[i])
      {
    dataPoints_[j]->setOperation(dpd_->getOperationFor(i));
    jsonDpIndex_.push_back(j);
    hasJson[j]=true;
    notFoundVar=false;
    break;
      }
    }
    if (notFoundVar) {
      assert(!strictChecking_);
      //push dummy DP if not registered by the service so that we output required JSON/CSV
      DataPoint *dummyDp = new DataPoint(sourceInfo_,defPath_);
      dummyDp->trackDummy(jsonNames[i],true);
      dataPoints_.push_back(dummyDp);
      jsonDpIndex_.push_back(dataPoints_.size()-1);
    }
  }
  for (unsigned int i=0;i<regDpCount_;i++) {
    dataPoints_[i]->setStreamLumiPtr(streamLumisPtr);
  }

  //fast path:
  if (haveFastPath_) {
    std::vector<std::string> const& fjsonNames = dpdFast_->getNames();
    fregDpCount_ = dataPointsFastOnly_.size();
    assert(!(fastPathStrictChecking_ && fjsonNames.size()==fregDpCount_));
    std::map<unsigned int,bool> fhasJson;
    for (unsigned int i=0;i<fjsonNames.size();i++)
    {
      bool notFoundVar=true;
      for (unsigned int j=0;j<fregDpCount_;j++) {
    if (dataPointsFastOnly_[j]->getName()==fjsonNames[i])
    {
      jsonDpIndexFast_.push_back(dataPointsFastOnly_[j]);
      fhasJson[j]=true;
      notFoundVar=false;
      break;
    }
      }
      if (notFoundVar)
      {
        //try to find variable among slow variables

        bool notFoundVarSlow=true;
        for (unsigned int j=0;j<regDpCount_;j++) {
      if (dataPoints_[j]->getName()==fjsonNames[i])
      {
        jsonDpIndexFast_.push_back(dataPoints_[j]);
        //fhasJson[j]=true;
        notFoundVarSlow=false;
        break;
      }
    }

    assert(!(fastPathStrictChecking_ && !notFoundVarSlow));
    //push dummy DP if not registered by the service so that we output required JSON/CSV
    if (notFoundVarSlow) {
      DataPoint *dummyDp = new DataPoint(sourceInfo_,defPathFast_);
      dummyDp->trackDummy(fjsonNames[i],true);
      dataPointsFastOnly_.push_back(dummyDp);
      jsonDpIndexFast_.push_back(dummyDp);
        }
      }
    }
  } 
}

//update everything
void FastMonitor::snap(unsigned int ls)
{
  recentSnaps_++;
  recentSnapsTimer_++;
  for (unsigned int i=0;i<regDpCount_;i++) {
    dataPoints_[i]->snap(ls);
  }
}

//update for global variables as most of them are correct only at global EOL
void FastMonitor::snapGlobal(unsigned int ls)
{
  recentSnaps_++;
  for (unsigned int i=0;i<regDpCount_;i++) {
    dataPoints_[i]->snapGlobal(ls);
  }
}

//update atomic per-stream vars(e.g. event counters) not updating time-based measurements (mini/microstate)
void FastMonitor::snapStreamAtomic(unsigned int ls, unsigned int streamID)
{
  recentSnaps_++;
  for (unsigned int i=0;i<regDpCount_;i++) {
    dataPoints_[i]->snapStreamAtomic(ls, streamID);
  }
}

std::string FastMonitor::getCSVString(int sid)
{
  //output what was specified in JSON in the same order (including dummies)
  unsigned int monSize = jsonDpIndexFast_.size();
  std::stringstream ss;
  if (monSize) {
    for (unsigned int j=0; j< monSize;j++) { 
      ss << jsonDpIndexFast_[j]->fastOutCSV(sid);
      if (j<monSize-1) ss << ",";
    }
  }
  return ss.str();
}

void FastMonitor::outputCSV(std::string const& path, std::string const& csvString)
{
  std::ofstream outputFile;
  outputFile.open(path.c_str(), std::fstream::out | std::fstream::trunc);
  outputFile << defPathFast_ << std::endl;
  outputFile << csvString << std::endl;
  outputFile.close();
}


//get one variable (caller must delete it later)
JsonMonitorable* FastMonitor::getMergedIntJForLumi(std::string const& name,unsigned int forLumi)
{
  auto it = dpNameMap_.find(name);
  assert(it!=dpNameMap_.end());
  return  dataPoints_[it->second]->mergeAndRetrieveValue(forLumi);
}

bool FastMonitor::outputFullJSONs(std::string const& pathstem, std::string const& ext, unsigned int lumi)
{
  LogDebug("FastMonitor") << "SNAP updates -: " <<  recentSnaps_ << " (by timer: " << recentSnapsTimer_ 
                          << ") in lumisection ";

  recentSnaps_ = recentSnapsTimer_ = 0;
  for (unsigned int i=0;i<nStreams_;i++) {
    Json::Value serializeRoot;
    for (unsigned int j=0; j< jsonDpIndex_.size();j++) {
      dataPoints_[jsonDpIndex_[j]]->mergeAndSerialize(serializeRoot,lumi,true,i);
    }
    //get extension
    std::stringstream tidext;
    tidext << "_tid" << i;
    std::string path = pathstem + tidext.str() + ext;

    Json::StyledWriter writer;
    std::string && result = writer.write(serializeRoot);
    FileIO::writeStringToFile(path, result);
  }
  return true;
}

bool FastMonitor::outputFullJSON(std::string const& path, unsigned int lumi)
{
  LogDebug("FastMonitor") << "SNAP updates -: " <<  recentSnaps_ << " (by timer: " << recentSnapsTimer_ 
                          << ") in lumisection ";

  recentSnaps_ = recentSnapsTimer_ = 0;
  Json::Value serializeRoot;
  for (unsigned int j=0; j< jsonDpIndex_.size();j++) {
    dataPoints_[jsonDpIndex_[j]]->mergeAndSerialize(serializeRoot,lumi,j==0,-1);
  }

  Json::StyledWriter writer;
  std::string && result = writer.write(serializeRoot);
  FileIO::writeStringToFile(path, result);
  return true;
}

void FastMonitor::discardCollected(unsigned int forLumi)
{
  for (auto dp: dataPoints_) dp->discardCollected(forLumi);
}

void FastMonitor::getHostAndPID(std::string& sHPid)
{
  std::stringstream hpid;
  int pid = (int) getpid();
  char hostname[128];
  gethostname(hostname, sizeof hostname);
  hpid << hostname << "_" << pid;
  sHPid = hpid.str();
}