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>;
}  // namespace jsoncollector

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 const *monitorable, bool NAifZeroUpdates) {
  name_ = monitorable->getName();
  tracked_ = (void const *)monitorable;
  if (dynamic_cast<IntJ const *>(monitorable))
    monType_ = TYPEINT;
  else if (dynamic_cast<DoubleJ const *>(monitorable))
    monType_ = TYPEDOUBLE;
  else if (dynamic_cast<StringJ const *>(monitorable))
    monType_ = TYPESTRING;
  else
    assert(0);
  NAifZeroUpdates_ = NAifZeroUpdates;
}

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

void DataPoint::trackVectorUIntAtomic(std::string const &name,
                                      std::vector<AtomicMonUInt *> const *monvec,
                                      bool NAifZeroUpdates) {
  name_ = name;
  tracked_ = (void const *)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 *> const *>(tracked_))->at(i)->load(std::memory_order_relaxed);
#else
        if (isAtomic_)
          monVal = *((static_cast<std::vector<AtomicMonUInt *> const *>(tracked_))->at(i));
#endif
        else
          monVal = (static_cast<std::vector<unsigned int> const *>(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 const *>(tracked_))->value());
      globalDataMap_[lumi] = ij;
    }
    if (monType_ == TYPEDOUBLE) {
      DoubleJ *dj = new DoubleJ;
      dj->update((static_cast<DoubleJ const *>(tracked_))->value());
      globalDataMap_[lumi] = dj;
    }
    if (monType_ == TYPESTRING) {
      StringJ *sj = new StringJ;
      sj->update((static_cast<StringJ const *>(tracked_))->value());
      globalDataMap_[lumi] = sj;
    }
  } else {
    if (monType_ == TYPEINT)
      static_cast<IntJ *>(itr->second.get())->update((static_cast<IntJ const *>(tracked_))->value());
    else if (monType_ == TYPEDOUBLE)
      static_cast<DoubleJ *>(itr->second.get())->update((static_cast<DoubleJ const *>(tracked_))->value());
    else if (monType_ == TYPESTRING)
      static_cast<StringJ *>(itr->second.get())->concatenate((static_cast<StringJ const *>(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 *> const *>(tracked_))->at(streamID)->load(std::memory_order_relaxed);
#else
    if (isAtomic_)
      monVal = *((static_cast<std::vector<AtomicMonUInt *> const *>(tracked_))->at(streamID));
#endif
    else
      monVal = (static_cast<std::vector<unsigned int> const *>(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 *> const *>(tracked_))->at(fastIndex_));
          fastIndex_ = (fastIndex_ + 1) % (static_cast<std::vector<AtomicMonUInt *> const *>(tracked_))->size();
        } else {
          ss << (static_cast<std::vector<unsigned int> const *>(tracked_))->at(fastIndex_);
          fastIndex_ = (fastIndex_ + 1) % (static_cast<std::vector<unsigned int> const *>(tracked_))->size();
        }

      } else {
        if (isAtomic_)
          ss << (unsigned int)*((static_cast<std::vector<AtomicMonUInt *> const *>(tracked_))->at(unsigned(sid)));
        else
          ss << (static_cast<std::vector<unsigned int> const *>(tracked_))->at(unsigned(sid));
      }
      return ss.str();
    }
    return (static_cast<JsonMonitorable const *>(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);
}