DataPoint.cc

Go to the documentation of this file.

/*
 * DataPoint.cc
 *
 *  Created on: Sep 24, 2012
 *      Author: aspataru
 */

#include "EventFilter/Utilities/interface/DataPoint.h"

#include <tbb/concurrent_vector.h>

#include <algorithm>
#include <cassert>

//max collected updates per lumi
#define MAXUPDATES 0xffffffff
#define MAXBINS

using namespace jsoncollector;

namespace jsoncollector {
  template class HistoJ<unsigned int>;
  template class HistoJ<double>;
}

const std::string DataPoint::SOURCE = "source";
const std::string DataPoint::DEFINITION = "definition";
const std::string DataPoint::DATA = "data";


DataPoint::~DataPoint()
{
  if (buf_) delete[] buf_;
}

/*
 *
 * Method implementation for simple DataPoint usage
 *
 */


void DataPoint::serialize(Json::Value& root) const
{
  if (!source_.empty()) {
    root[SOURCE] = source_;
  }
  if (!definition_.empty()) {
    root[DEFINITION] = definition_;
  }
  for (unsigned int i=0;i<data_.size();i++)
    root[DATA].append(data_[i]);
}


void DataPoint::deserialize(Json::Value& root)
{
  source_ = root.get(SOURCE, "").asString();
  definition_ = root.get(DEFINITION, "").asString();
  if (root.get(DATA, "").isArray()) {
    unsigned int size = root.get(DATA, "").size();
    for (unsigned int i = 0; i < size; i++) {
      data_.push_back(root.get(DATA, "")[i].asString());
    }
  }
}

/*
 *
 * Method implementation for the new multi-threaded model
 *
 * */

void DataPoint::trackMonitorable(JsonMonitorable *monitorable,bool NAifZeroUpdates)
{
  name_=monitorable->getName();
  tracked_ = (void*)monitorable;
  if (dynamic_cast<IntJ*>(monitorable)) monType_=TYPEINT;
  else if (dynamic_cast<DoubleJ*>(monitorable)) monType_=TYPEDOUBLE;
  else if (dynamic_cast<StringJ*>(monitorable)) monType_=TYPESTRING;
  else assert(0);
  NAifZeroUpdates_=NAifZeroUpdates;

}

void DataPoint::trackVectorUInt(std::string const& name, std::vector<unsigned int>  *monvec, bool NAifZeroUpdates)
{
  name_=name;
  tracked_ = (void*)monvec;
  isStream_=true;
  monType_=TYPEUINT;
  NAifZeroUpdates_=NAifZeroUpdates;
  makeStreamLumiMap(monvec->size());
}

void DataPoint::trackVectorUIntAtomic(std::string const& name, std::vector<AtomicMonUInt*>  *monvec, bool NAifZeroUpdates)
{
  name_=name;
  tracked_ = (void*)monvec;
  isStream_=true;
  isAtomic_=true;
  monType_=TYPEUINT;
  NAifZeroUpdates_=NAifZeroUpdates;
  makeStreamLumiMap(monvec->size());
}

void DataPoint::makeStreamLumiMap(unsigned int size)
{
  for (unsigned int i=0;i<size;i++) {
    streamDataMaps_.push_back(MonPtrMap());
  }
}

void DataPoint::serialize(Json::Value& root, bool rootInit, std::string const&input) const
{
  if (rootInit) {
    if (!source_.empty())
      root[SOURCE] = source_;
    if (!definition_.empty())
      root[DEFINITION] = definition_;
  }
  root[DATA].append(input);
}

void DataPoint::snap(unsigned int lumi)
{
  isCached_=false;
  if (isStream_) {
    if (monType_==TYPEUINT)
    {
      for (unsigned int i=0; i<streamDataMaps_.size();i++) {
        unsigned int streamLumi_=streamLumisPtr_->at(i);//get currently processed stream lumi
        unsigned int monVal;

#if ATOMIC_LEVEL>0
        if (isAtomic_) monVal = (static_cast<std::vector<AtomicMonUInt*>*>(tracked_))->at(i)->load(std::memory_order_relaxed);
#else 
        if (isAtomic_) monVal = *((static_cast<std::vector<AtomicMonUInt*>*>(tracked_))->at(i));
#endif
    else monVal = (static_cast<std::vector<unsigned int>*>(tracked_))->at(i);

    auto itr =  streamDataMaps_[i].find(streamLumi_);
    if (itr==streamDataMaps_[i].end())
    {
      if (opType_==OPHISTO) {
            if (*nBinsPtr_) {
              HistoJ<unsigned int> *nh = new HistoJ<unsigned int>(1,MAXUPDATES);
              nh->update(monVal);
              streamDataMaps_[i][streamLumi_] = nh;
            }
          }
          else {//default to SUM
            IntJ *nj = new IntJ;
            nj->update(monVal);
            streamDataMaps_[i][streamLumi_]= nj;
          }
        }
        else { 
          if (opType_==OPHISTO) {
            if (*nBinsPtr_) {
              (static_cast<HistoJ<unsigned int> *>(itr->second.get()))->update(monVal);
            }
          }
          else {
            *(static_cast<IntJ*>(itr->second.get()))=monVal;
          }
        }
      }
    }
    else assert(monType_!=TYPEINT);//not yet implemented, application error
  }
  else snapGlobal(lumi);
}

void DataPoint::snapGlobal(unsigned int lumi)
{
  isCached_=false;
  if (isStream_) return;
  auto itr = globalDataMap_.find(lumi);
  if (itr==globalDataMap_.end()) {
    if (monType_==TYPEINT) {
      IntJ *ij = new IntJ;
      ij->update((static_cast<IntJ*>(tracked_))->value());
      globalDataMap_[lumi]=ij;
    }
    if (monType_==TYPEDOUBLE) {
      DoubleJ *dj = new DoubleJ;
      dj->update((static_cast<DoubleJ*>(tracked_))->value());
      globalDataMap_[lumi]=dj;
    }
    if (monType_==TYPESTRING) {
      StringJ *sj = new StringJ;
      sj->update((static_cast<StringJ*>(tracked_))->value());
      globalDataMap_[lumi]=sj;
    }
  } else { 
    if (monType_==TYPEINT)
      static_cast<IntJ*>(itr->second.get())->update((static_cast<IntJ*>(tracked_))->value());
    else if (monType_==TYPEDOUBLE)
      static_cast<DoubleJ*>(itr->second.get())->update((static_cast<DoubleJ*>(tracked_))->value());
    else if (monType_==TYPESTRING)
      static_cast<StringJ*>(itr->second.get())->concatenate((static_cast<StringJ*>(tracked_))->value());
  }
}

void DataPoint::snapStreamAtomic(unsigned int lumi, unsigned int streamID)
{
  if (!isStream_ || !isAtomic_) return;
  isCached_=false;
  if (monType_==TYPEUINT)
  {
    unsigned int monVal;
#if ATOMIC_LEVEL>0
    if (isAtomic_) monVal = (static_cast<std::vector<AtomicMonUInt*>*>(tracked_))->at(streamID)->load(std::memory_order_relaxed);
#else 
    if (isAtomic_) monVal = *((static_cast<std::vector<AtomicMonUInt*>*>(tracked_))->at(streamID));
#endif
    else monVal = (static_cast<std::vector<unsigned int>*>(tracked_))->at(streamID);

    auto itr =  streamDataMaps_[streamID].find(lumi);
    if (itr==streamDataMaps_[streamID].end()) //insert
    {
      if (opType_==OPHISTO) {
        if (*nBinsPtr_) {
          HistoJ<unsigned int> *h = new HistoJ<unsigned int>(1,MAXUPDATES);
          h->update(monVal);
          streamDataMaps_[streamID][lumi] = h;
        }
      }
      else {//default to SUM

        IntJ *h = new IntJ;
        h->update(monVal);
        streamDataMaps_[streamID][lumi] = h;
      }
    }
    else 
    { 
      if (opType_==OPHISTO) {
        if (*nBinsPtr_) {
          static_cast<HistoJ<unsigned int>*>(itr->second.get())->update(monVal);
        }
      }
      else
        *(static_cast<IntJ*>(itr->second.get()))=monVal;
    }
  }
  else assert(monType_!=TYPEINT);//not yet implemented
}

std::string DataPoint::fastOutCSV(int sid)
{
  if (tracked_) {
    if (isStream_) {
      std::stringstream ss;
      if (sid<0) {
        if (isAtomic_) { 
//        if ATOMIC_LEVEL>0
//        ss << (unsigned int) (static_cast<std::vector<AtomicMonUInt*>*>(tracked_))->at(fastIndex_)->load(std::memory_order_relaxed); 

          ss << (unsigned int) *((static_cast<std::vector<AtomicMonUInt*>*>(tracked_))->at(fastIndex_));
          fastIndex_ = (fastIndex_+1) % (static_cast<std::vector<AtomicMonUInt*>*>(tracked_))->size();
        }
        else {
          ss << (static_cast<std::vector<unsigned int>*>(tracked_))->at(fastIndex_);
          fastIndex_ = (fastIndex_+1) % (static_cast<std::vector<unsigned int>*>(tracked_))->size();
        }

      }
      else {
        if (isAtomic_)
          ss << (unsigned int) *((static_cast<std::vector<AtomicMonUInt*>*>(tracked_))->at(unsigned(sid)));
        else
          ss << (static_cast<std::vector<unsigned int>*>(tracked_))->at(unsigned(sid));
      }
      return ss.str();
    }
    return (static_cast<JsonMonitorable*>(tracked_))->toString();
  }
  return std::string("");
}

JsonMonitorable* DataPoint::mergeAndRetrieveValue(unsigned int lumi)
{
  assert(monType_==TYPEUINT && isStream_);//for now only support UINT and SUM for stream variables
  IntJ *newJ = new IntJ;
  for (unsigned int i=0;i<streamDataMaps_.size();i++) {
    auto itr = streamDataMaps_[i].find(lumi);
    if (itr!=streamDataMaps_[i].end()) {
      newJ->add(static_cast<IntJ*>(itr->second.get())->value());
    }
  }
  cacheI_=newJ->value();
  isCached_=true;
  return newJ;//assume the caller takes care of deleting the object
}

void DataPoint::mergeAndSerialize(Json::Value & root,unsigned int lumi,bool initJsonValue, int sid)
{
  if (initJsonValue) {
    root[SOURCE] = source_;
    root[DEFINITION] = definition_;
  }

  if (isDummy_) {
    root[DATA].append("N/A");
    return;
  }
  if (!isStream_) {
    auto itr = globalDataMap_.find(lumi);
    if (itr != globalDataMap_.end()) {
      root[DATA].append(itr->second.get()->toString());
    }
    else {
      if (NAifZeroUpdates_) root[DATA].append("N/A");
      else if (monType_==TYPESTRING)  root[DATA].append("");
      else  root[DATA].append("0");
    }
    return;
  }
  else {
    assert(monType_==TYPEUINT);
    if (isCached_) {
      std::stringstream ss;
      ss << cacheI_;
      root[DATA].append(ss.str());
      return;
    }
    if (opType_!=OPHISTO) {//sum is default
      std::stringstream ss;
      unsigned int updates=0;
      unsigned int sum=0;
      if (sid<1)
        for (unsigned int i=0;i<streamDataMaps_.size();i++) {
          auto itr = streamDataMaps_[i].find(lumi);
          if (itr!=streamDataMaps_[i].end()) {
            sum+=static_cast<IntJ*>(itr->second.get())->value();
            updates++;
          }
        }
      else {
        auto itr = streamDataMaps_[unsigned(sid)].find(lumi);
        if (itr!=streamDataMaps_[unsigned(sid)].end()) {
          sum+=static_cast<IntJ*>(itr->second.get())->value();
          updates++;
        }
      }
      if (!updates && NAifZeroUpdates_) ss << "N/A";
      ss << sum;
      root[DATA].append(ss.str());
      return;
    }
    if (opType_==OPHISTO) {
      if (nBinsPtr_==nullptr) {
        root[DATA].append("N/A");
        return;
      }
      if (*nBinsPtr_>bufLen_) {
        if (buf_) delete[] buf_;
        bufLen_=*nBinsPtr_;
        buf_= new uint32_t[bufLen_];
      }
      memset(buf_,0,bufLen_*sizeof(uint32_t));
      unsigned int updates=0;
      if (sid<1)
      for (unsigned int i=0;i<streamDataMaps_.size();i++) {
          auto itr = streamDataMaps_[i].find(lumi);
          if (itr!=streamDataMaps_[i].end()) {
            HistoJ <unsigned int>* monObj = static_cast<HistoJ<unsigned int>*>(itr->second.get());
            updates+=monObj->getUpdates();
            auto &hvec = monObj->value();
            for (unsigned int j=0;j<hvec.size();j++) {
              unsigned int thisbin=(unsigned int) hvec[j];
              if (thisbin<*nBinsPtr_) {
                buf_[thisbin]++;
              }
            }
          }
        }
      else {
        auto itr = streamDataMaps_[unsigned(sid)].find(lumi);
        if (itr!=streamDataMaps_[unsigned(sid)].end()) {
          HistoJ <unsigned int>* monObj = static_cast<HistoJ<unsigned int>*>(itr->second.get());
          updates+=monObj->getUpdates();
          auto &hvec = monObj->value();
          for (unsigned int j=0;j<hvec.size();j++) {
            unsigned int thisbin=(unsigned int) hvec[j];
            if (thisbin<*nBinsPtr_) {
              buf_[thisbin]++;
            }
          }
        }
      }
      std::stringstream ss;
      if (!*nBinsPtr_ || (!updates && NAifZeroUpdates_)) ss << "N/A";
      else {
        ss << "[";
        if (*nBinsPtr_) {
          for (unsigned int i=0;i<*nBinsPtr_-1;i++) {
            ss << buf_[i] << ",";
          }
          ss << buf_[*nBinsPtr_-1];
        }
        ss << "]";
      }
      root[DATA].append(ss.str());
      return;
    }
  }
}

//wipe out data that will no longer be used
void DataPoint::discardCollected(unsigned int lumi)
{
  for (unsigned int i=0;i<streamDataMaps_.size();i++)
  {
    auto itr = streamDataMaps_[i].find(lumi);
    if (itr!=streamDataMaps_[i].end()) streamDataMaps_[i].erase(lumi);
  }

  auto itr = globalDataMap_.find(lumi);
  if (itr!=globalDataMap_.end())
    globalDataMap_.erase(lumi);
}