CbmDeviceUnpack.h

Go to the documentation of this file.

/* Copyright (C) 2021 Facility for Antiproton and Ion Research in Europe, Darmstadt
   SPDX-License-Identifier: GPL-3.0-only
   Authors: Pierre-Alain Loizeau [committer] */
 
#ifndef CBMDEVICEUNPACK_H_
#define CBMDEVICEUNPACK_H_
 
#include "CbmMqTMessage.h"
#include "CbmTsEventHeader.h"
 
#include "Timeslice.hpp"
 
#include "FairMQDevice.h"
#include "FairParGenericSet.h"
 
#include "Rtypes.h"
#include "TObjArray.h"
 
#include <chrono>
#include <map>
#include <vector>
 
class TList;
class CbmBmonUnpackConfig;
class CbmStsUnpackConfig;
class CbmMuchUnpackConfig;
class CbmTrdUnpackFaspConfig;
class CbmTrdUnpackConfig;
class CbmTofUnpackConfig;
class CbmRichUnpackConfig;
class CbmPsdUnpackConfig;
 
class TimesliceMetaData;
 
class CbmTrdSpadic;
 
class CbmDeviceUnpack : public FairMQDevice {
public:
  CbmDeviceUnpack();
  virtual ~CbmDeviceUnpack();
 
protected:
  virtual void InitTask();
  bool ConditionalRun();
  bool HandleCommand(FairMQMessagePtr&, int);
 
  void SetUnpackConfig(std::shared_ptr<CbmBmonUnpackConfig> config) { fBmonConfig = config; }
 
  void SetUnpackConfig(std::shared_ptr<CbmStsUnpackConfig> config) { fStsConfig = config; }
 
  void SetUnpackConfig(std::shared_ptr<CbmMuchUnpackConfig> config) { fMuchConfig = config; }
 
  void SetUnpackConfig(std::shared_ptr<CbmTrdUnpackConfig> config) { fTrd1DConfig = config; }
 
  void SetUnpackConfig(std::shared_ptr<CbmTrdUnpackFaspConfig> config) { fTrd2DConfig = config; }
 
  void SetUnpackConfig(std::shared_ptr<CbmTofUnpackConfig> config) { fTofConfig = config; }
 
  void SetUnpackConfig(std::shared_ptr<CbmRichUnpackConfig> config) { fRichConfig = config; }
 
  void SetUnpackConfig(std::shared_ptr<CbmPsdUnpackConfig> config) { fPsdConfig = config; }
 
private:
  static constexpr std::uint16_t fkFlesBmon  = static_cast<std::uint16_t>(fles::Subsystem::BMON);
  static constexpr std::uint16_t fkFlesMvd   = static_cast<std::uint16_t>(fles::Subsystem::MVD);
  static constexpr std::uint16_t fkFlesSts   = static_cast<std::uint16_t>(fles::Subsystem::STS);
  static constexpr std::uint16_t fkFlesMuch  = static_cast<std::uint16_t>(fles::Subsystem::MUCH);
  static constexpr std::uint16_t fkFlesTrd   = static_cast<std::uint16_t>(fles::Subsystem::TRD);
  static constexpr std::uint16_t fkFlesTrd2D = static_cast<std::uint16_t>(fles::Subsystem::TRD2D);
  static constexpr std::uint16_t fkFlesTof   = static_cast<std::uint16_t>(fles::Subsystem::TOF);
  static constexpr std::uint16_t fkFlesRich  = static_cast<std::uint16_t>(fles::Subsystem::RICH);
  static constexpr std::uint16_t fkFlesPsd   = static_cast<std::uint16_t>(fles::Subsystem::PSD);
 
 
  Bool_t fbIgnoreOverlapMs       = false;  
  Bool_t fbComponentsAddedToList = kFALSE;
 
  bool fDoDebugPrints = false;  
  bool fDoPerfProf = false;  
  bool fbOutputFullTimeSorting = false;
 
  std::string fsSetupName = "mcbm_beam_2021_07_surveyed";
  uint32_t fuRunId        = 1588;
  bool fbUnpBmon  = false;
  bool fbUnpSts   = true;
  bool fbUnpMuch  = false;
  bool fbUnpTrd1D = true;
  bool fbUnpTrd2D = true;
  bool fbUnpTof   = true;
  bool fbUnpRich  = true;
  bool fbUnpPsd   = true;
  std::string fsChannelNameDataInput   = "ts-request";
  std::string fsChannelNameDataOutput  = "unpts_0";
  std::string fsChannelNameCommands    = "commands";
  std::string fsChannelNameHistosInput = "histogram-in";
  uint32_t fuPublishFreqTs  = 100;
  double_t fdMinPublishTime = 0.5;
  double_t fdMaxPublishTime = 5.0;
 
  //      TList* fParCList = nullptr;
  Bool_t InitParameters(std::vector<std::pair<std::string, std::shared_ptr<FairParGenericSet>>>* reqparvec);
 
  uint64_t fulNumMessages                                = 0;
  uint64_t fulTsCounter                                  = 0;
  std::chrono::system_clock::time_point fLastPublishTime = std::chrono::system_clock::now();
  std::map<std::uint16_t, std::pair<std::string, size_t>> fNameMap = {};  
  std::map<std::uint16_t, std::pair<double, double>> fTimeMap = {};  
  std::map<std::uint16_t, std::pair<double, double>> fDataSizeMap = {};  
 
  std::shared_ptr<CbmBmonUnpackConfig> fBmonConfig     = nullptr;
  std::shared_ptr<CbmStsUnpackConfig> fStsConfig       = nullptr;
  std::shared_ptr<CbmMuchUnpackConfig> fMuchConfig     = nullptr;
  std::shared_ptr<CbmTrdUnpackFaspConfig> fTrd2DConfig = nullptr;
  std::shared_ptr<CbmTrdUnpackConfig> fTrd1DConfig     = nullptr;
  std::shared_ptr<CbmTofUnpackConfig> fTofConfig       = nullptr;
  std::shared_ptr<CbmRichUnpackConfig> fRichConfig     = nullptr;
  std::shared_ptr<CbmPsdUnpackConfig> fPsdConfig       = nullptr;
 
  CbmTsEventHeader* fCbmTsEventHeader = nullptr;
 
  std::vector<std::string> fvsSetTimeOffs = {};
 
  size_t fuNbCoreMsPerTs    = 0;     
  size_t fuNbOverMsPerTs    = 0;     
  Double_t fdMsSizeInNs     = 0;     
  Double_t fdTsCoreSizeInNs = -1.0;  
  Double_t fdTsOverSizeInNs = -1.0;  
  Double_t fdTsFullSizeInNs = -1.0;  
  TimesliceMetaData* fTsMetaData;
 
  TObjArray fArrayHisto = {};
  std::vector<std::pair<std::string, std::string>> fvpsHistosFolder = {};
  std::vector<std::pair<std::string, std::string>> fvpsCanvasConfig = {};
  bool fbConfigSent = false;
 
  Bool_t InitContainers();
  bool InitHistograms();
  Bool_t DoUnpack(const fles::Timeslice& ts, size_t component);
  void Finish();
  bool SendUnpData();
  bool SendHistoConfAndData();
  bool SendHistograms();
 
  std::shared_ptr<CbmTrdSpadic> GetTrdSpadic(bool useAvgBaseline);
 
  template<typename TVecobj>
  typename std::enable_if<std::is_same<TVecobj, std::nullptr_t>::value == true, void>::type
  timesort(std::vector<TVecobj>* /*vec = nullptr*/)
  {
    LOG(debug) << "CbmDeviceUnpack::timesort() got an object that has no member function GetTime(). Hence, we can and "
                  "will not timesort it!";
  }
 
  template<typename TVecobj>
  typename std::enable_if<!std::is_member_function_pointer<decltype(&TVecobj::GetTime)>::value, void>::type
  timesort(std::vector<TVecobj>* /*vec = nullptr*/)
  {
    LOG(debug) << "CbmDeviceUnpack::timesort() " << TVecobj::Class_Name()
               << "is  an object that has no member function GetTime(). Hence, we can and "
                  "will not timesort it!";
  }
 
  template<typename TVecobj>
  typename std::enable_if<std::is_member_function_pointer<decltype(&TVecobj::GetTime)>::value, void>::type
  timesort(std::vector<TVecobj>* vec = nullptr)
  {
    if (vec == nullptr) return;
    std::sort(vec->begin(), vec->end(),
              [](const TVecobj& a, const TVecobj& b) -> bool { return a.GetTime() < b.GetTime(); });
  }
 
  template<class TConfig, class TOptOutA = std::nullptr_t, class TOptOutB = std::nullptr_t>
  size_t unpack(const std::uint16_t subsysid, const fles::Timeslice* ts, std::uint16_t icomp, TConfig config,
                std::vector<TOptOutA>* optouttargetvecA = nullptr, std::vector<TOptOutB>* optouttargetvecB = nullptr)
  {
 
    auto wallstarttime        = std::chrono::high_resolution_clock::now();
    std::clock_t cpustarttime = std::clock();
 
    auto algo                        = config->GetUnpacker();
    std::vector<TOptOutA> optoutAvec = {};
    std::vector<TOptOutB> optoutBvec = {};
    if (optouttargetvecA) { algo->SetOptOutAVec(&optoutAvec); }
    if (optouttargetvecB) { algo->SetOptOutBVec(&optoutBvec); }
 
    // Set the start time of the current TS for this algorithm
    algo->SetTsStartTime(ts->start_time());
 
    // Run the actual unpacking
    auto digivec = algo->Unpack(ts, icomp);
 
    // Check if we want to write the output to somewhere (in pure online monitoring mode for example this can/would/should be skipped)
    if (config->GetOutputVec()) {
      // Lets do some time-sorting if we are not doing it later
      if (!fbOutputFullTimeSorting) timesort(&digivec);
 
      // Transfer the data from the timeslice vector to the target branch vector
      // Digis/default output retrieved as offered by the algorithm
      for (auto digi : digivec)
        config->GetOutputVec()->emplace_back(digi);
    }
    if (optouttargetvecA) {
      // Lets do some timesorting
      if (!fbOutputFullTimeSorting) timesort(&optoutAvec);
      // Transfer the data from the timeslice vector to the target branch vector
      for (auto optoutA : optoutAvec)
        optouttargetvecA->emplace_back(optoutA);
    }
    if (optouttargetvecB) {
      // Second opt output is not time sorted to allow non GetTime data container.
      // Lets do some timesorting
      timesort(&optoutAvec);
      // Transfer the data from the timeslice vector to the target branch vector
      for (auto optoutB : optoutBvec)
        optouttargetvecB->emplace_back(optoutB);
    }
 
    std::clock_t cpuendtime = std::clock();
    auto wallendtime        = std::chrono::high_resolution_clock::now();
 
    // Cpu time in [µs]
    auto cputime = 1e6 * (cpuendtime - cpustarttime) / CLOCKS_PER_SEC;
    algo->AddCpuTime(cputime);
    // Real time in [µs]
    auto walltime = std::chrono::duration<double, std::micro>(wallendtime - wallstarttime).count();
    algo->AddWallTime(walltime);
 
 
    // Check some numbers from this timeslice
    size_t nDigis = digivec.size();
    LOG(debug) << "Component " << icomp << " connected to config " << config->GetName() << "   n-Digis " << nDigis
               << " processed in walltime(cputime) = " << walltime << "(" << cputime << cputime << ") µs"
               << "this timeslice.";
 
    if (fDoPerfProf) {
      auto timeit = fTimeMap.find(subsysid);
      timeit->second.first += cputime;
      timeit->second.second += walltime;
 
      auto datait = fDataSizeMap.find(subsysid);
      datait->second.first += ts->size_component(icomp) / 1.0e6;
      datait->second.second += nDigis * algo->GetOutputObjSize() / 1.0e6;
 
      fNameMap.find(subsysid)->second.second += nDigis;
    }
 
    return nDigis;
  }
};
 
#endif /* CBMDEVICEMCBMUNPACK_H_ */