CMVToVectorSpec.cxx

Go to the documentation of this file.

// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
 
 
#include <limits>
#include <memory>
#include <stdexcept>
#include <vector>
#include <string>
#include <fstream>
#include <algorithm>
#include <fmt/format.h>
#include <fmt/chrono.h>
 
#include "TFile.h"
#include "DetectorsRaw/RDHUtils.h"
#include "Framework/Task.h"
#include "Framework/ConfigParamRegistry.h"
#include "Framework/Logger.h"
#include "Framework/DataProcessorSpec.h"
#include "Framework/WorkflowSpec.h"
#include "Framework/InputRecordWalker.h"
#include "Framework/DataRefUtils.h"
#include "DPLUtils/RawParser.h"
#include "Headers/DataHeader.h"
#include "Headers/DataHeaderHelpers.h"
#include "CommonUtils/TreeStreamRedirector.h"
 
#include "DataFormatsTPC/CMV.h"
#include "DataFormatsTPC/RawDataTypes.h"
#include "TPCBase/RDHUtils.h"
#include "TPCBase/Mapper.h"
#include "TPCWorkflow/ProcessingHelpers.h"
 
using namespace o2::framework;
using o2::header::gDataOriginTPC;
using RDHUtils = o2::raw::RDHUtils;
using RawDataType = o2::tpc::raw_data_types::Type;
 
namespace o2::tpc
{
 
class CMVToVectorDevice : public o2::framework::Task
{
 public:
  using FEEIDType = rdh_utils::FEEIDType;
  CMVToVectorDevice(const std::vector<uint32_t>& crus) : mCRUs(crus) {}
 
  void init(o2::framework::InitContext& ic) final
  {
    // set up ADC value filling
    mWriteDebug = ic.options().get<bool>("write-debug");
    mWriteDebugOnError = ic.options().get<bool>("write-debug-on-error");
    mWriteRawDataOnError = ic.options().get<bool>("write-raw-data-on-error");
    mRawDataType = ic.options().get<int>("raw-data-type");
    o2::framework::RawParser<>::setCheckIncompleteHBF(ic.options().get<bool>("check-incomplete-hbf"));
 
    mDebugStreamFileName = ic.options().get<std::string>("debug-file-name").data();
    mRawOutputFileName = ic.options().get<std::string>("raw-file-name").data();
 
    initCMV();
  }
 
  void run(o2::framework::ProcessingContext& pc) final
  {
    const auto runNumber = processing_helpers::getRunNumber(pc);
    std::vector<InputSpec> filter = {{"check", ConcreteDataTypeMatcher{o2::header::gDataOriginTPC, "RAWDATA"}, Lifetime::Timeframe}};
 
    // open files if necessary
    if ((mWriteDebug || mWriteDebugOnError) && !mDebugStream) {
      const auto debugFileName = fmt::format(fmt::runtime(mDebugStreamFileName), fmt::arg("run", runNumber));
      LOGP(info, "Creating debug stream {}", debugFileName);
      mDebugStream = std::make_unique<o2::utils::TreeStreamRedirector>(debugFileName.data(), "recreate");
    }
 
    if (mWriteRawDataOnError && !mRawOutputFile.is_open()) {
      std::string_view rawType = (mRawDataType < 2) ? "tf" : "raw";
      if (mRawDataType == 5) {
        rawType = "cmv.raw";
      }
      const auto rawFileName = fmt::format(fmt::runtime(mRawOutputFileName), fmt::arg("run", runNumber), fmt::arg("raw_type", rawType));
      LOGP(info, "Creating raw debug file {}", rawFileName);
      mRawOutputFile.open(rawFileName, std::ios::binary);
    }
 
    uint32_t tfCounter = 0;
    bool hasErrors = false;
 
    for (auto const& ref : InputRecordWalker(pc.inputs(), filter)) {
      const auto* dh = DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
      tfCounter = dh->tfCounter;
      const auto subSpecification = dh->subSpecification;
      auto payloadSize = DataRefUtils::getPayloadSize(ref);
      LOGP(debug, "Processing TF {}, subSpecification {}, payloadSize {}", tfCounter, subSpecification, payloadSize);
 
      // ---| data loop |---
      const gsl::span<const char> raw = pc.inputs().get<gsl::span<char>>(ref);
      try {
        o2::framework::RawParser parser(raw.data(), raw.size());
        size_t lastErrorCount = 0;
 
        for (auto it = parser.begin(), end = parser.end(); it != end; ++it) {
          const auto size = it.size();
 
          if (parser.getNErrors() > lastErrorCount) {
            lastErrorCount = parser.getNErrors();
            hasErrors = true;
          }
 
          // skip empty packages (HBF open)
          if (size == 0) {
            continue;
          }
 
          auto rdhPtr = reinterpret_cast<const o2::header::RDHAny*>(it.raw());
          const auto rdhVersion = RDHUtils::getVersion(rdhPtr);
          if (!rdhPtr || rdhVersion < 6) {
            throw std::runtime_error(fmt::format("could not get RDH from packet, or version {} < 6", rdhVersion).data());
          }
 
          // ---| extract hardware information to do the processing |---
          const auto feeId = (FEEIDType)RDHUtils::getFEEID(*rdhPtr);
          const auto link = rdh_utils::getLink(feeId);
          const uint32_t cruID = rdh_utils::getCRU(feeId);
          const auto detField = RDHUtils::getDetectorField(*rdhPtr);
 
          LOGP(debug, "Detected CMV packet: CRU {}, link {}, feeId {}", cruID, link, feeId);
 
          if ((detField != (decltype(detField))RawDataType::CMV) || (link != rdh_utils::CMVLinkID)) {
            LOGP(debug, "Skipping packet: detField {}, (expected RawDataType {}), link {}, (expected CMVLinkID {})", detField, (decltype(detField))RawDataType::CMV, link, rdh_utils::CMVLinkID);
            continue;
          }
 
          LOGP(debug, "Processing firstTForbit {:9}, tfCounter {:5}, run {:6}, feeId {:6}, cruID {:3}, link {:2}", dh->firstTForbit, dh->tfCounter, dh->runNumber, feeId, cruID, link);
 
          if (std::find(mCRUs.begin(), mCRUs.end(), cruID) == mCRUs.end()) {
            // LOGP(debug, "CMV CRU {:3} not configured in CRUs, skipping", cruID);
            continue;
          }
 
          auto& cmvVec = mCMVvectors[cruID];
          auto& infoVec = mCMVInfos[cruID];
 
          if (size != sizeof(cmv::Container)) {
            LOGP(warning, "CMV packet size mismatch: got {} bytes, expected {} bytes (sizeof cmv::Container). Skipping package.", size, sizeof(cmv::Container));
            hasErrors = true;
            continue;
          }
          auto data = it.data();
          auto& cmvs = *((cmv::Container*)(data));
          const uint32_t orbit = cmvs.header.heartbeatOrbit;
          const uint16_t bc = cmvs.header.heartbeatBC;
 
          // record packet meta and append its CMV vector (3564 TB)
          infoVec.emplace_back(orbit, bc);
          cmvVec.reserve(cmvVec.size() + cmv::NTimeBinsPerPacket);
          for (uint32_t tb = 0; tb < cmv::NTimeBinsPerPacket; ++tb) {
            cmvVec.push_back(cmvs.getCMV(tb));
            // LOGP(debug, "For CRU {}, timebin {}, orbit {}, bc {}, appended CMV {} float: {}", cruID, tb, orbit, bc, cmvs.getCMV(tb), cmvs.getCMVFloat(tb));
          }
        }
      } catch (const std::exception& e) {
        // error message throtteling
        using namespace std::literals::chrono_literals;
        static std::unordered_map<uint32_t, size_t> nErrorPerSubspec;
        static std::chrono::time_point<std::chrono::steady_clock> lastReport = std::chrono::steady_clock::now();
        const auto now = std::chrono::steady_clock::now();
        static size_t reportedErrors = 0;
        const size_t MAXERRORS = 10;
        const auto sleepTime = 10min;
        ++nErrorPerSubspec[subSpecification];
 
        if ((now - lastReport) < sleepTime) {
          if (reportedErrors < MAXERRORS) {
            ++reportedErrors;
            std::string sleepInfo;
            if (reportedErrors == MAXERRORS) {
              sleepInfo = fmt::format(", maximum error count ({}) reached, not reporting for the next {}", MAXERRORS, sleepTime);
            }
            LOGP(alarm, "EXCEPTION in processRawData: {} -> skipping part:{}/{} of spec:{}/{}/{}, size:{}, error count for subspec: {}{}", e.what(), dh->splitPayloadIndex, dh->splitPayloadParts,
                 dh->dataOrigin, dh->dataDescription, subSpecification, payloadSize, nErrorPerSubspec.at(subSpecification), sleepInfo);
            lastReport = now;
          }
        } else {
          lastReport = now;
          reportedErrors = 0;
        }
        continue;
      }
    }
 
    hasErrors |= snapshotCMVs(pc.outputs());
 
    if (mWriteDebug || (mWriteDebugOnError && hasErrors)) {
      writeDebugOutput(tfCounter);
    }
 
    if (mWriteRawDataOnError && hasErrors) {
      writeRawData(pc.inputs());
    }
 
    // clear output
    initCMV();
  }
 
  void closeFiles()
  {
    LOGP(info, "closeFiles");
 
    if (mDebugStream) {
      // set some default aliases
      auto& stream = (*mDebugStream) << "cmvs";
      auto& tree = stream.getTree();
      tree.SetAlias("sector", "int(cru/10)");
      mDebugStream->Close();
      mDebugStream.reset(nullptr);
      mRawOutputFile.close();
    }
  }
 
  void stop() final
  {
    LOGP(info, "stop");
    closeFiles();
  }
 
  void endOfStream(o2::framework::EndOfStreamContext& ec) final
  {
    LOGP(info, "endOfStream");
    // ec.services().get<ControlService>().readyToQuit(QuitRequest::Me);
    closeFiles();
  }
 
 private:
  struct CMVInfo {
    CMVInfo() = default;
    CMVInfo(const CMVInfo&) = default;
    CMVInfo(uint32_t orbit, uint16_t bc) : heartbeatOrbit(orbit), heartbeatBC(bc) {}
 
    uint32_t heartbeatOrbit{0};
    uint16_t heartbeatBC{0};
 
    bool operator==(const uint32_t orbit) const { return (heartbeatOrbit == orbit); }
    bool operator==(const CMVInfo& inf) const { return (inf.heartbeatOrbit == heartbeatOrbit) && (inf.heartbeatBC == heartbeatBC); }
    bool matches(uint32_t orbit, int16_t bc) const { return ((heartbeatOrbit == orbit) && (heartbeatBC == bc)); }
  };
 
  int mRawDataType{0};                                             
  bool mWriteDebug{false};                                         
  bool mWriteDebugOnError{false};                                  
  bool mWriteRawDataOnError{false};                                
  std::vector<uint32_t> mCRUs;                                     
  std::unordered_map<uint32_t, std::vector<uint16_t>> mCMVvectors; 
  std::unordered_map<uint32_t, std::vector<CMVInfo>> mCMVInfos;    
  std::string mDebugStreamFileName;                                
  std::unique_ptr<o2::utils::TreeStreamRedirector> mDebugStream;   
  std::ofstream mRawOutputFile;                                    
  std::string mRawOutputFileName;                                  
 
  //____________________________________________________________________________
  bool snapshotCMVs(DataAllocator& output)
  {
    bool hasErrors = false;
 
    // send data per CRU with its own orbit/BC vector
    for (auto& [cru, cmvVec] : mCMVvectors) {
      const header::DataHeader::SubSpecificationType subSpec{cru << 7};
      const auto& infVec = mCMVInfos[cru];
 
      if (infVec.size() != 4) {
        // LOGP(error, "CRU {:3}: expected 4 packets per TF, got {}", cru, infVec.size());
        hasErrors = true;
      }
      if (cmvVec.size() != cmv::NTimeBinsPerPacket * infVec.size()) {
        // LOGP(error, "CRU {:3}: vector size {} does not match expected {}", cru, cmvVec.size(), cmv::NTimeBinsPerPacket * infVec.size());
        hasErrors = true;
      }
 
      std::vector<uint64_t> orbitBCInfo;
      orbitBCInfo.reserve(infVec.size());
      for (const auto& inf : infVec) {
        orbitBCInfo.emplace_back((uint64_t(inf.heartbeatOrbit) << 32) + uint64_t(inf.heartbeatBC));
      }
 
      LOGP(debug, "Sending CMVs for CRU {} of size {} ({} packets)", cru, cmvVec.size(), infVec.size());
      output.snapshot(Output{gDataOriginTPC, "CMVVECTOR", subSpec}, cmvVec);
      output.snapshot(Output{gDataOriginTPC, "CMVORBITS", subSpec}, orbitBCInfo);
    }
 
    return hasErrors;
  }
 
  //____________________________________________________________________________
  void initCMV()
  {
    for (const auto cruID : mCRUs) {
      auto& cmvVec = mCMVvectors[cruID];
      cmvVec.clear();
 
      auto& infosCRU = mCMVInfos[cruID];
      infosCRU.clear();
    }
  }
 
  //____________________________________________________________________________
  void writeDebugOutput(uint32_t tfCounter)
  {
    mDebugStream->GetFile()->cd();
    auto& stream = (*mDebugStream) << "cmvs";
 
    for (auto cru : mCRUs) {
      if (mCMVInfos.find(cru) == mCMVInfos.end()) {
        continue;
      }
 
      auto& infos = mCMVInfos[cru];
      auto& cmvVec = mCMVvectors[cru];
 
      stream << "cru=" << cru
             << "tfCounter=" << tfCounter
             << "nCMVs=" << cmvVec.size()
             << "cmvs=" << cmvVec
             << "\n";
    }
  }
 
  void writeRawData(InputRecord& inputs)
  {
    if (!mRawOutputFile.is_open()) {
      return;
    }
 
    using DataHeader = o2::header::DataHeader;
 
    std::vector<InputSpec> filter = {{"check", ConcreteDataTypeMatcher{o2::header::gDataOriginTPC, "RAWDATA"}, Lifetime::Timeframe}};
    for (auto const& ref : InputRecordWalker(inputs, filter)) {
      auto dh = DataRefUtils::getHeader<header::DataHeader*>(ref);
      // LOGP(info, "write header: {}/{}/{}, payload size: {} / {}", dh->dataOrigin, dh->dataDescription, dh->subSpecification, dh->payloadSize, ref.payloadSize);
      if (((mRawDataType == 1) || (mRawDataType == 3)) && (dh->payloadSize == 2 * sizeof(o2::header::RAWDataHeader))) {
        continue;
      }
 
      if (mRawDataType < 2) {
        mRawOutputFile.write(ref.header, sizeof(DataHeader));
      }
      if (mRawDataType < 5) {
        mRawOutputFile.write(ref.payload, ref.payloadSize);
      }
 
      if (mRawDataType == 5) {
        const gsl::span<const char> raw = inputs.get<gsl::span<char>>(ref);
        try {
          o2::framework::RawParser parser(raw.data(), raw.size());
          for (auto it = parser.begin(), end = parser.end(); it != end; ++it) {
            const auto size = it.size();
            // skip empty packages (HBF open)
            if (size == 0) {
              continue;
            }
 
            auto rdhPtr = reinterpret_cast<const o2::header::RDHAny*>(it.raw());
            const auto rdhVersion = RDHUtils::getVersion(rdhPtr);
            if (!rdhPtr || rdhVersion < 6) {
              throw std::runtime_error(fmt::format("could not get RDH from packet, or version {} < 6", rdhVersion).data());
            }
 
            // ---| extract hardware information to do the processing |---
            const auto feeId = (FEEIDType)RDHUtils::getFEEID(*rdhPtr);
            const auto link = rdh_utils::getLink(feeId);
            const auto detField = RDHUtils::getDetectorField(*rdhPtr);
 
            // only select CMVs
            if ((detField != (decltype(detField))RawDataType::CMV) || (link != rdh_utils::CMVLinkID)) {
              continue;
            }
 
            // write out raw data
            mRawOutputFile.write((const char*)it.raw(), RDHUtils::getMemorySize(rdhPtr));
          }
        } catch (...) {
        }
      }
    }
  }
};
 
o2::framework::DataProcessorSpec getCMVToVectorSpec(std::string const& inputSpec, std::vector<uint32_t> const& crus)
{
  using device = o2::tpc::CMVToVectorDevice;
 
  std::vector<OutputSpec> outputs;
  for (const uint32_t cru : crus) {
    const header::DataHeader::SubSpecificationType subSpec{cru << 7};
    outputs.emplace_back(gDataOriginTPC, "CMVVECTOR", subSpec, Lifetime::Timeframe);
    outputs.emplace_back(gDataOriginTPC, "CMVORBITS", subSpec, Lifetime::Timeframe);
  }
 
  return DataProcessorSpec{
    fmt::format("tpc-cmv-to-vector"),
    select(inputSpec.data()),
    outputs,
    AlgorithmSpec{adaptFromTask<device>(crus)},
    Options{
      {"write-debug", VariantType::Bool, false, {"write a debug output tree"}},
      {"write-debug-on-error", VariantType::Bool, false, {"write a debug output tree in case errors occurred"}},
      {"debug-file-name", VariantType::String, "/tmp/cmv_vector_debug.{run}.root", {"name of the debug output file"}},
      {"write-raw-data-on-error", VariantType::Bool, false, {"dump raw data in case errors occurred"}},
      {"raw-file-name", VariantType::String, "/tmp/cmv_debug.{run}.{raw_type}", {"name of the raw output file"}},
      {"raw-data-type", VariantType::Int, 0, {"Which raw data to dump: 0-full TPC with DH, 1-full TPC with DH skip empty, 2-full TPC no DH, 3-full TPC no DH skip empty, 4-IDC raw only 5-CMV raw only"}},
      {"check-incomplete-hbf", VariantType::Bool, false, {"false: don't check; true: check and report"}},
    } // end Options
  }; // end DataProcessorSpec
}
} // namespace o2::tpc