CbmMQTsSamplerRepReq.cxx

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] */
 
#include "CbmMQTsSamplerRepReq.h"
 
#include "CbmFlesCanvasTools.h"
#include "CbmFormatDecHexPrintout.h"
 
#include "TimesliceInputArchive.hpp"
#include "TimesliceMultiInputArchive.hpp"
#include "TimesliceMultiSubscriber.hpp"
#include "TimesliceSubscriber.hpp"
 
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"  // device->fConfig
 
#include <TCanvas.h>
#include <TH1F.h>
#include <TH1I.h>
#include <TProfile.h>
 
#include "BoostSerializer.h"
#include <boost/algorithm/string.hpp>
#include <boost/archive/binary_oarchive.hpp>
//#include <boost/filesystem.hpp>
#include <boost/regex.hpp>
#include <boost/serialization/utility.hpp>
 
#include "RootSerializer.h"
 
//namespace filesys = boost::filesystem;
 
#include <algorithm>
#include <chrono>
#include <cstdio>
#include <ctime>
#include <iomanip>
#include <sstream>
#include <string>
#include <thread>  // this_thread::sleep_for
 
using namespace std;
 
#include <stdexcept>
 
struct InitTaskError : std::runtime_error {
  using std::runtime_error::runtime_error;
};
 
CbmMQTsSamplerRepReq::CbmMQTsSamplerRepReq()
  : FairMQDevice()
  , fTime()
  , fLastPublishTime {std::chrono::system_clock::now()}
{
}
 
void CbmMQTsSamplerRepReq::InitTask()
try {
  // Get the values from the command line options (via fConfig)
  fsFileName               = fConfig->GetValue<string>("filename");
  fsDirName                = fConfig->GetValue<string>("dirname");
  fsHost                   = fConfig->GetValue<string>("fles-host");
  fusPort                  = fConfig->GetValue<uint16_t>("fles-port");
  fulHighWaterMark         = fConfig->GetValue<uint64_t>("high-water-mark");
  fulMaxTimeslices         = fConfig->GetValue<uint64_t>("max-timeslices");
  fsChannelNameTsRequest   = fConfig->GetValue<std::string>("ChNameTsReq");
  fbNoSplitTs              = fConfig->GetValue<bool>("no-split-ts");
  fbSendTsPerSysId         = fConfig->GetValue<bool>("send-ts-per-sysid");
  fbSendTsPerBlock         = fConfig->GetValue<bool>("send-ts-per-block");
  fsChannelNameMissedTs    = fConfig->GetValue<std::string>("ChNameMissTs");
  fsChannelNameCommands    = fConfig->GetValue<std::string>("ChNameCmds");
  fuPublishFreqTs          = fConfig->GetValue<uint32_t>("PubFreqTs");
  fdMinPublishTime         = fConfig->GetValue<double_t>("PubTimeMin");
  fdMaxPublishTime         = fConfig->GetValue<double_t>("PubTimeMax");
  fsHistosSuffix           = fConfig->GetValue<std::string>("HistosSuffix");
  fsChannelNameHistosInput = fConfig->GetValue<std::string>("ChNameIn");
 
  if (fbNoSplitTs) {
    if (fbSendTsPerSysId) {
      if (fbSendTsPerBlock) {
        LOG(warning) << "Both no-split-ts, send-ts-per-sysid and send-ts-per-block options used => "
                     << " second and third one will be ignored!!!!";
      }  // if( fbSendTsPerBlock )
      else
        LOG(warning) << "Both no-split-ts and send-ts-per-sysid options used => "
                     << " second one will be ignored!!!!";
    }  // if( fbSendTsPerSysId )
    else if (fbSendTsPerBlock) {
      LOG(warning) << "Both no-split-ts and send-ts-per-block options used => "
                   << " second one will be ignored!!!!";
    }  // else if( fbSendTsPerSysId ) of if( fbSendTsPerSysId )
    else
      LOG(debug) << "Running in no-split-ts mode!";
  }  // if( fbNoSplitTs )
  else if (fbSendTsPerBlock) {
    if (fbSendTsPerSysId) {
      LOG(warning) << "Both send-ts-per-sysid and send-ts-per-block options used => "
                   << " second one will be ignored!!!!";
    }  // if (fbSendTsPerSysId)
    else
      LOG(debug) << "Running in send-ts-per-block mode!";
  }  // else if( fbSendTsPerBlock ) of if( fbNoSplitTs )
  else if (fbSendTsPerSysId) {
    LOG(debug) << "Running in send-ts-per-sysid mode!";
  }  // else if (fbSendTsPerSysId) else if( fbSendTsPerBlock ) of if( fbNoSplitTs )
  else {
    LOG(debug) << "Running in no-split-ts mode by default!";
    fbNoSplitTs = true;
  }  // else of else if (fbSendTsPerSysId) else if( fbSendTsPerBlock ) of if( fbNoSplitTs )

  std::vector<std::string> vSysIdBlockPairs = fConfig->GetValue<std::vector<std::string>>("block-sysid");
  for (uint32_t uPair = 0; uPair < vSysIdBlockPairs.size(); ++uPair) {
    const size_t sep = vSysIdBlockPairs[uPair].find(':');
    if (string::npos == sep || 0 == sep || vSysIdBlockPairs[uPair].size() == sep) {
      LOG(info) << vSysIdBlockPairs[uPair];
      throw InitTaskError("Provided pair of Block name + SysId is missing a : or an argument.");
    }  // if( string::npos == sep || 0 == sep || vSysIdBlockPairs[ uPair ].size() == sep )

    std::string sBlockName = vSysIdBlockPairs[uPair].substr(0, sep);

    std::string sSysId  = vSysIdBlockPairs[uPair].substr(sep + 1);
    const size_t hexPos = sSysId.find("0x");
    uint16_t usSysId;
    if (string::npos == hexPos) usSysId = std::stoi(sSysId);
    else
      usSysId = std::stoi(sSysId.substr(hexPos + 2), nullptr, 16);
 
    LOG(debug) << "Extracted block info from pair \"" << vSysIdBlockPairs[uPair] << "\": name is " << sBlockName
               << " and SysId is " << usSysId << " extracted from " << sSysId;

    uint32_t uSysIdIdx = 0;
    for (; uSysIdIdx < fSysId.size() && fSysId[uSysIdIdx] != usSysId; ++uSysIdIdx) {}
    if (uSysIdIdx == fSysId.size()) { throw InitTaskError("Unknown System ID for " + vSysIdBlockPairs[uPair]); }
    else if (true == fComponentActive[uSysIdIdx]) {
      throw InitTaskError("System ID already in use by another block for " + vSysIdBlockPairs[uPair]);
    }
    fComponentActive[uSysIdIdx] = true;

    auto itBlock = fvBlocksToSend.begin();
    for (; itBlock != fvBlocksToSend.end(); ++itBlock) {
      if ((*itBlock).first == sBlockName) break;
    }  // for( ; itBlock != fvBlocksToSend.end(); ++itBlock)
    if (fvBlocksToSend.end() != itBlock) {
      (*itBlock).second.insert(usSysId);
    }  // if( fvBlocksToSend.end() != itBlock )
    else {
      fvBlocksToSend.push_back(std::pair<std::string, std::set<uint16_t>>(sBlockName, {usSysId}));
      fvvCompPerBlock.push_back(std::vector<uint32_t>({}));
    }  // else of if( fSysId.end() != pos )
 
    LOG(info) << vSysIdBlockPairs[uPair] << " Added SysId 0x" << std::hex << usSysId << std::dec << " to "
              << sBlockName;
  }  // for( uint32_t uPair = 0; uPair < vSysIdBlockPairs.size(); ++uPair )
 
  if (0 == fulMaxTimeslices) fulMaxTimeslices = UINT_MAX;
 
  // Check which input is defined
  // Possibilities:
  // filename && ! dirname : single file
  // filename with wildcards && dirname : all files with filename regex in the directory
  // host && port : connect to the flesnet server
  bool isGoodInputCombi {false};
  if (0 != fsFileName.size() && 0 == fsDirName.size() && 0 == fsHost.size() && 0 == fusPort) {
    isGoodInputCombi = true;
    fvsInputFileList.push_back(fsFileName);
  }
  else if (0 != fsFileName.size() && 0 != fsDirName.size() && 0 == fsHost.size() && 0 == fusPort) {
    isGoodInputCombi = true;
    fvsInputFileList.push_back(fsFileName);
  }
  else if (0 == fsFileName.size() && 0 == fsDirName.size() && 0 != fsHost.size() && 0 != fusPort) {
    isGoodInputCombi = true;
    LOG(info) << "Host: " << fsHost;
    LOG(info) << "Port: " << fusPort;
  }
  else if (0 == fsFileName.size() && 0 == fsDirName.size() && 0 != fsHost.size() && 0 == fusPort) {
    isGoodInputCombi = true;
    LOG(info) << "Host string: " << fsHost;
  }
  else {
    isGoodInputCombi = false;
  }
 
  if (!isGoodInputCombi) {
    throw InitTaskError("Wrong combination of inputs. Either file or wildcard file + directory "
                        "or host + port are allowed combination.");
  }
 
  LOG(info) << "MaxTimeslices: " << fulMaxTimeslices;
 
  if (0 == fsFileName.size() && 0 != fsHost.size() && 0 != fusPort) {
    // Don't add the protocol since this is done now in the TimesliceMultiSubscriber
    //std::string connector = "tcp://" + fsHost + ":" + std::to_string(fusPort);
    std::string connector = fsHost + ":" + std::to_string(fusPort);
    LOG(info) << "Open TSPublisher at " << connector;
    fSource = new fles::TimesliceMultiSubscriber(connector, fulHighWaterMark);
  }
  else if (0 == fsFileName.size() && 0 != fsHost.size()) {
    std::string connector = fsHost;
    LOG(info) << "Open TSPublisher with host string: " << connector;
    fSource = new fles::TimesliceMultiSubscriber(connector, fulHighWaterMark);
  }
  else {
    // Create a ";" separated string with all file names
    std::string fileList {""};
    for (const auto& obj : fvsInputFileList) {
      std::string fileName = obj;
      fileList += fileName;
      fileList += ";";
    }
    fileList.pop_back();  // Remove the last ;
    LOG(info) << "Input File String: " << fileList;
    fSource = new fles::TimesliceMultiInputArchive(fileList, fsDirName);
    if (!fSource) { throw InitTaskError("Could open files from file list."); }
  }
 
  LOG(info) << "High-Water Mark: " << fulHighWaterMark;
  LOG(info) << "Max. Timeslices: " << fulMaxTimeslices;
  if (fbNoSplitTs) { LOG(info) << "Sending TS copies in no-split mode"; }  // if( fbNoSplitTs )
  else if (fbSendTsPerSysId) {
    LOG(info) << "Sending components in separate TS per SysId";
  }  // else if( fbSendTsPerSysId ) of if( fbNoSplitTs )
  else if (fbSendTsPerBlock) {
    LOG(info) << "Sending components in separate TS per block (multiple SysId)";
  }  // else if( fbSendTsPerBlock ) of if( fbSendTsPerSysId ) of if( fbNoSplitTs )
 
  OnData(fsChannelNameTsRequest, &CbmMQTsSamplerRepReq::HandleRequest);
 
  fTime = std::chrono::steady_clock::now();
}
catch (InitTaskError& e) {
  LOG(error) << e.what();
  ChangeState(fair::mq::Transition::ErrorFound);
}
catch (boost::bad_any_cast& e) {
  LOG(error) << "Error during InitTask: " << e.what();
  ChangeState(fair::mq::Transition::ErrorFound);
}
 
bool CbmMQTsSamplerRepReq::InitHistograms()
{
  LOG(info) << "Histograms publication frequency in TS:    " << fuPublishFreqTs;
  LOG(info) << "Histograms publication min. interval in s: " << fdMinPublishTime;
  LOG(info) << "Histograms publication max. interval in s: " << fdMaxPublishTime;
  if ("" != fsHistosSuffix) {  //
    LOG(info) << "Suffix added to folders, histograms and canvas names: " << fsHistosSuffix;
  }

  std::vector<std::pair<TNamed*, std::string>> vHistos = {};
  std::vector<std::pair<TCanvas*, std::string>> vCanvases = {};

  /* clang-format off */
  fhTsRate       = new TH1I(Form("TsRate%s", fsHistosSuffix.data()),
                            "TS rate; t [s]",
                            1800, 0., 1800.);
  fhTsSize       = new TH1I(Form("TsSize%s", fsHistosSuffix.data()),
                           "Size of TS; Size [MB]",
                           15000, 0., 15000.);
  fhTsSizeEvo    = new TProfile(Form("TsSizeEvo%s", fsHistosSuffix.data()),
                                "Evolution of the TS Size; t [s]; Mean size [MB]",
                                1800, 0., 1800.);
  fhTsMaxSizeEvo = new TH1F(Form("TsMaxSizeEvo%s", fsHistosSuffix.data()),
                            "Evolution of maximal TS Size; t [s]; Max size [MB]",
                            1800, 0., 1800.);
  fhMissedTS     = new TH1I(Form("MissedTs%s", fsHistosSuffix.data()),
                            "Missed TS",
                            2, -0.5, 1.5);
  fhMissedTSEvo  = new TProfile(Form("MissedTsEvo%s", fsHistosSuffix.data()),
                                "Missed TS evolution; t [s]",
                                1800, 0., 1800.);
  /* clang-format on */

  std::string sFolder = std::string("Sampler") + fsHistosSuffix;
  vHistos.push_back(std::pair<TNamed*, std::string>(fhTsRate, sFolder));
  vHistos.push_back(std::pair<TNamed*, std::string>(fhTsSize, sFolder));
  vHistos.push_back(std::pair<TNamed*, std::string>(fhTsSizeEvo, sFolder));
  vHistos.push_back(std::pair<TNamed*, std::string>(fhTsMaxSizeEvo, sFolder));
  vHistos.push_back(std::pair<TNamed*, std::string>(fhMissedTS, sFolder));
  vHistos.push_back(std::pair<TNamed*, std::string>(fhMissedTSEvo, sFolder));

  Double_t w = 10;
  Double_t h = 10;
  fcSummary  = new TCanvas(Form("cSampSummary%s", fsHistosSuffix.data()), "Sampler monitoring plots", w, h);
  fcSummary->Divide(2, 3);
 
  fcSummary->cd(1);
  gPad->SetGridx();
  gPad->SetGridy();
  fhTsRate->Draw("hist");
 
  fcSummary->cd(2);
  gPad->SetGridx();
  gPad->SetGridy();
  gPad->SetLogx();
  gPad->SetLogy();
  fhTsSize->Draw("hist");
 
  fcSummary->cd(3);
  gPad->SetGridx();
  gPad->SetGridy();
  fhTsSizeEvo->Draw("hist");
 
  fcSummary->cd(4);
  gPad->SetGridx();
  gPad->SetGridy();
  fhTsMaxSizeEvo->Draw("hist");
 
  fcSummary->cd(5);
  gPad->SetGridx();
  gPad->SetGridy();
  fhMissedTS->Draw("hist");
 
  fcSummary->cd(6);
  gPad->SetGridx();
  gPad->SetGridy();
  fhMissedTSEvo->Draw("el");

  vCanvases.push_back(std::pair<TCanvas*, std::string>(fcSummary, std::string("canvases") + fsHistosSuffix));

  for (UInt_t uHisto = 0; uHisto < vHistos.size(); ++uHisto) {
    //         LOG(info) << "Registering  " << vHistos[ uHisto ].first->GetName()
    //                   << " in " << vHistos[ uHisto ].second.data()
    //                   ;
    fArrayHisto.Add(vHistos[uHisto].first);
    std::pair<std::string, std::string> psHistoConfig(vHistos[uHisto].first->GetName(), vHistos[uHisto].second);
    fvpsHistosFolder.push_back(psHistoConfig);
 
    LOG(info) << "Config of hist  " << psHistoConfig.first.data() << " in folder " << psHistoConfig.second.data();
  }  // for( UInt_t uHisto = 0; uHisto < vHistos.size(); ++uHisto )

  for (UInt_t uCanv = 0; uCanv < vCanvases.size(); ++uCanv) {
    //         LOG(info) << "Registering  " << vCanvases[ uCanv ].first->GetName()
    //                   << " in " << vCanvases[ uCanv ].second.data();
    std::string sCanvName = (vCanvases[uCanv].first)->GetName();
    std::string sCanvConf = GenerateCanvasConfigString(vCanvases[uCanv].first);
 
    std::pair<std::string, std::string> psCanvConfig(sCanvName, sCanvConf);
 
    fvpsCanvasConfig.push_back(psCanvConfig);
 
    LOG(info) << "Config string of Canvas  " << psCanvConfig.first.data() << " is " << psCanvConfig.second.data();
  }  //  for( UInt_t uCanv = 0; uCanv < vCanvases.size(); ++uCanv )
 
  return true;
}
 
bool CbmMQTsSamplerRepReq::HandleRequest(FairMQMessagePtr& msgReq, int)
{
  if (0 < fuPublishFreqTs && 0 == fulTsCounter) { InitHistograms(); }  // if( 0 < fuPublishFreqTs )
 
  if (fbEofFound) {
    return true;
  }

  std::string reqStr(static_cast<char*>(msgReq->GetData()), msgReq->GetSize());
  if ("SendFirstTimesliceIndex" == reqStr) {
    if (0 == fulFirstTsIndex) {  //
      GetNewTs();
    }
    if (!SendFirstTsIndex() && !fbEofFound) {  //
      return false;
    }
    return true;
  }
 
  if (fbNoSplitTs) {
 
    if (!CreateAndSendFullTs() && !fbEofFound) {
      if ("" != fsChannelNameCommands) {
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        SendCommand("STOP");
      }  // if( "" != fsChannelNameCommands )
 
      return false;
    }  // if( !CreateAndSendFullTs( ts ) && !fbEofFound)
  }    // if( fbNoSplitTs )
  else if (fbSendTsPerSysId) {
    int iSysId = std::stoi(reqStr);
    LOG(debug) << "Received TS SysId component request from client: 0x" << std::hex << iSysId << std::dec;

    if (!CreateCombinedComponentsPerSysId(iSysId) && !fbEofFound) {
      if ("" != fsChannelNameCommands) {
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        SendCommand("STOP");
      }  // if( "" != fsChannelNameCommands )
 
      return false;
    }  // if(!CreateAndCombineComponentsPerSysId(iSysId) && !fbEofFound)
  }    // else if( fbSendTsPerSysId && fbSendTsPerSysId ) of if( fbNoSplitTs
  else if (fbSendTsPerBlock) {
    LOG(debug) << "Received TS components block request from client: " << reqStr;

    if (!CreateCombinedComponentsPerBlock(reqStr) && !fbEofFound) {
      if ("" != fsChannelNameCommands) {
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        SendCommand("STOP");
      }  // if( "" != fsChannelNameCommands )
 
      return false;
    }  // if( !CreateAndCombineComponentsPerChannel(reqStr) && !fbEofFound)
  }    // else if( fbSendTsPerSysId && fbSendTsPerSysId ) of if( fbNoSplitTs )

  std::chrono::system_clock::time_point currentTime = std::chrono::system_clock::now();
  std::chrono::duration<double_t> elapsedSeconds    = currentTime - fLastPublishTime;
  if ((fdMaxPublishTime < elapsedSeconds.count())
      || (0 == fulMessageCounter % fuPublishFreqTs && fdMinPublishTime < elapsedSeconds.count())) {
    if (!fbConfigSent) {
      // Send the configuration only once per run!
      fbConfigSent = SendHistoConfAndData();
    }  // if( !fbConfigSent )
    else
      SendHistograms();
 
    fLastPublishTime = std::chrono::system_clock::now();
  }  // if( ( fdMaxPublishTime < elapsedSeconds.count() ) || ( 0 == fulMessageCounter % fuPublishFreqTs && fdMinPublishTime < elapsedSeconds.count() ) )
 
  return true;
}
 
std::unique_ptr<fles::Timeslice> CbmMQTsSamplerRepReq::GetNewTs()
{
  if (0 == fulTsCounter && nullptr != dynamic_cast<fles::TimesliceMultiSubscriber*>(fSource)) {
    dynamic_cast<fles::TimesliceMultiSubscriber*>(fSource)->InitTimesliceSubscriber();
  }  // if( 0 == fulTsCounter && nullptr != dynamic_cast< fles::TimesliceMultiSubscriber >(fSource) )
 
  std::unique_ptr<fles::Timeslice> timeslice = fSource->get();
  if (timeslice) {
    if (fulTsCounter < fulMaxTimeslices) {
 
      const fles::Timeslice& ts = *timeslice;
      uint64_t uTsIndex         = ts.index();
 
      if (0 == fulFirstTsIndex) {  //
        fulFirstTsIndex = ts.descriptor(0, 0).idx;
      }
 
      if (0 < fuPublishFreqTs) {
        uint64_t uTsTime = ts.descriptor(0, 0).idx;
        if (0 == fuStartTime) {  //
          fuStartTime = uTsTime;
        }  // if( 0 == fuStartTime )
        fdTimeToStart    = static_cast<double_t>(uTsTime - fuStartTime) / 1e9;
        uint64_t uSizeMb = 0;
 
        for (uint64_t uComp = 0; uComp < ts.num_components(); ++uComp) {
          uSizeMb += ts.size_component(uComp) / (1024 * 1024);
        }  // for( uint_t uComp = 0; uComp < ts.num_components(); ++uComp )
 
 
        fhTsRate->Fill(fdTimeToStart);
        fhTsSize->Fill(uSizeMb);
        fhTsSizeEvo->Fill(fdTimeToStart, uSizeMb);

        if (0. == fdLastMaxTime) {
          fdLastMaxTime = fdTimeToStart;
          fdTsMaxSize   = uSizeMb;
        }  // if( 0. == fdLastMaxTime )
        else if (1. <= fdTimeToStart - fdLastMaxTime) {
          fhTsMaxSizeEvo->Fill(fdLastMaxTime, fdTsMaxSize);
          fdLastMaxTime = fdTimeToStart;
          fdTsMaxSize   = uSizeMb;
        }  // else if if( 1 <= fdTimeToStart - fdLastMaxTime )
        else if (fdTsMaxSize < uSizeMb) {
          fdTsMaxSize = uSizeMb;
        }  // else if( fdTsMaxSize < uSizeMb )
      }    // if( 0 < fuPublishFreqTs )

      if ((uTsIndex != (fulPrevTsIndex + 1)) && !(0 == fulPrevTsIndex && 0 == uTsIndex && 0 == fulTsCounter)) {
        LOG(debug) << "Missed Timeslices. Old TS Index was " << fulPrevTsIndex << " New TS Index is " << uTsIndex
                   << " diff is " << uTsIndex - fulPrevTsIndex << " Missing are " << uTsIndex - fulPrevTsIndex - 1;
 
        if ("" != fsChannelNameMissedTs) {
          std::vector<uint64_t> vulMissedIndices;
          if (0 == fulPrevTsIndex && 0 == fulTsCounter) {
            vulMissedIndices.emplace_back(0);
          }
          for (uint64_t ulMiss = fulPrevTsIndex + 1; ulMiss < uTsIndex; ++ulMiss) {
            vulMissedIndices.emplace_back(ulMiss);
          }  // for( uint64_t ulMiss = fulPrevTsIndex + 1; ulMiss < uTsIndex; ++ulMiss )
          if (!SendMissedTsIdx(vulMissedIndices)) {
            if ("" != fsChannelNameCommands) {
              std::this_thread::sleep_for(std::chrono::milliseconds(1000));
              SendCommand("STOP");
            }  // if( "" != fsChannelNameCommands )
 
            return nullptr;
          }  // if( !SendMissedTsIdx( vulMissedIndices ) )
        }    // if( "" != fsChannelNameMissedTs )
 
        if (0 < fuPublishFreqTs) {
          fhMissedTS->Fill(1, uTsIndex - fulPrevTsIndex - 1);
          fhMissedTSEvo->Fill(fdTimeToStart, 1, uTsIndex - fulPrevTsIndex - 1);
        }  // if( 0 < fuPublishFreqTs )
 
      }  // if( ( uTsIndex != ( fulPrevTsIndex + 1 ) ) && !( 0 == fulPrevTsIndex && 0 == uTsIndex ) )
 
      if (0 < fuPublishFreqTs) {
        fhMissedTS->Fill(0);
        fhMissedTSEvo->Fill(fdTimeToStart, 0, 1);
      }  // else if( 0 < fuPublishFreqTs )
 
      fulTsCounter++;
      fulPrevTsIndex = uTsIndex;
 
      if (fulTsCounter % 10000 == 0) { LOG(info) << "Received TS " << fulTsCounter << " with index " << uTsIndex; }
 
      LOG(debug) << "Found " << ts.num_components() << " different components in timeslice";
      return timeslice;
    }  // if (fulTsCounter < fulMaxTimeslices)
    else {
      CalcRuntime();

      if ("" != fsChannelNameCommands) {
        std::this_thread::sleep_for(std::chrono::seconds(10));
        std::string sCmd = "EOF ";
        sCmd += FormatDecPrintout(fulPrevTsIndex);
        sCmd += " ";
        sCmd += FormatDecPrintout(fulTsCounter);
        SendCommand(sCmd);
      }  // if( "" != fsChannelNameCommands )
 
      fbEofFound = true;
 
      return nullptr;
    }  // else of if (fulTsCounter < fulMaxTimeslices)
  }  // if (timeslice)
  else {
    CalcRuntime();

    if ("" != fsChannelNameCommands) {
      std::this_thread::sleep_for(std::chrono::seconds(10));
      std::string sCmd = "EOF ";
      sCmd += FormatDecPrintout(fulPrevTsIndex);
      sCmd += " ";
      sCmd += FormatDecPrintout(fulTsCounter);
      SendCommand(sCmd);
    }  // if( "" != fsChannelNameCommands )
 
    fbEofFound = true;
 
    return nullptr;
  }  // else of if (timeslice)
}
 
bool CbmMQTsSamplerRepReq::AddNewTsInBuffer()
{
  while (fulHighWaterMark <= fdpTimesliceBuffer.size()) {
    fdpTimesliceBuffer.pop_front();
    fdbCompSentFlags.pop_front();
  }  // while( fulHighWaterMark <= fdpTimesliceBuffer.size() )

  fdpTimesliceBuffer.push_back(GetNewTs());
  if (nullptr == fdpTimesliceBuffer.back()) {
    fdpTimesliceBuffer.pop_back();
    return false;
  }  // if(nullptr == fdpTimesliceBuffer[fdpTimesliceBuffer.size() - 1])

  if (fbSendTsPerBlock) { fdbCompSentFlags.push_back(std::vector<bool>(fvBlocksToSend.size(), false)); }
  else {
    fdbCompSentFlags.push_back(std::vector<bool>(fComponentActive.size(), false));
  }
  return true;
}
 
bool CbmMQTsSamplerRepReq::CreateAndSendFullTs()
{
  std::unique_ptr<fles::Timeslice> timeslice = GetNewTs();
  if (timeslice) {
    const fles::Timeslice& ts = *timeslice;
    fles::StorableTimeslice fullTs {ts};
    if (!SendData(fullTs)) {
      if ("" != fsChannelNameCommands) {
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        SendCommand("STOP");
      }  // if( "" != fsChannelNameCommands )
 
      return false;
    }  // if (!SendData(fullTs, uChanIdx))
    return true;
  }  // if (timeslice)
  else {
    return false;
  }  // else of if (timeslice)
}
 
bool CbmMQTsSamplerRepReq::PrepareCompListPerSysId()
{
  if (false == fbListCompPerSysIdReady) {
    if (0 == fdpTimesliceBuffer.size()) {
      if (!AddNewTsInBuffer()) return false;
    }  // if( 0 == fdpTimesliceBuffer.size() )
 
    if (nullptr == fdpTimesliceBuffer.front()) return false;
 
    for (uint32_t uCompIdx = 0; uCompIdx < fdpTimesliceBuffer.front()->num_components(); ++uCompIdx) {
      uint16_t usMsSysId = fdpTimesliceBuffer.front()->descriptor(uCompIdx, 0).sys_id;
 
      const vector<int>::const_iterator pos = std::find(fSysId.begin(), fSysId.end(), usMsSysId);
      if (fSysId.end() != pos) {
        const vector<std::string>::size_type idx = pos - fSysId.begin();
 
        fvvCompPerSysId[idx].push_back(uCompIdx);
      }  // if( fSysId.end() != pos )
    }    // for( uint32_t uCompIdx = 0; uCompIdx < ts.num_components(); ++uCompIdx )
 
    for (uint32_t uSysIdx = 0; uSysIdx < fComponents.size(); ++uSysIdx) {
      std::stringstream ss;
      ss << "Found " << std::setw(2) << fvvCompPerSysId[uSysIdx].size() << " components for SysId 0x" << std::hex
         << std::setw(2) << fSysId[uSysIdx] << std::dec << " :";
 
      for (uint32_t uComp = 0; uComp < fvvCompPerSysId[uSysIdx].size(); ++uComp) {
        ss << " " << std::setw(3) << fvvCompPerSysId[uSysIdx][uComp];
      }  // for( uint32_t uComp = 0; uComp < fvvCompPerSysId[ uSysIdx ].size(); ++uComp )
 
      LOG(info) << ss.str();
    }  // for( uint32_t uSysId = 0; uSysId < fSysId.size(); ++uSysId )
 
    fbListCompPerSysIdReady = true;
  }  // if( false == fbListCompPerSysIdReady )
 
  return true;
}
bool CbmMQTsSamplerRepReq::CreateCombinedComponentsPerSysId(std::string sSystemName)
{
  if (!PrepareCompListPerSysId()) return false;

  const vector<std::string>::const_iterator pos = std::find(fComponents.begin(), fComponents.end(), sSystemName);
  if (fComponents.end() != pos) {
    const vector<std::string>::size_type idx = pos - fComponents.begin();
    return CreateCombinedComponentsPerSysId(static_cast<uint32_t>(idx));
  }  // if (fComponents.end() != pos)
  else {
    LOG(error) << "Did not find " << sSystemName << " in the list of known systems";
    return false;
  }  // else of if (fComponents.end() != pos)
}
bool CbmMQTsSamplerRepReq::CreateCombinedComponentsPerSysId(int iSysId)
{
  if (!PrepareCompListPerSysId()) return false;

  const vector<int>::const_iterator pos = std::find(fSysId.begin(), fSysId.end(), iSysId);
  if (fSysId.end() != pos) {
    const vector<int>::size_type idx = pos - fSysId.begin();
    return CreateCombinedComponentsPerSysId(static_cast<uint32_t>(idx));
  }  // if (fSysId.end() != pos)
  else {
    LOG(error) << "Did not find 0x" << std::hex << iSysId << std::dec << " in the list of known systems";
    return false;
  }  // else of if (fSysId.end() != pos)
}
bool CbmMQTsSamplerRepReq::CreateCombinedComponentsPerSysId(uint uCompIndex)
{
  LOG(debug) << "Create timeslice with components for SysId " << std::hex << fSysId[uCompIndex] << std::dec;
 
  if (0 < fvvCompPerSysId[uCompIndex].size()) {
    uint32_t uTsIndex = 0;
    for (; uTsIndex < fdpTimesliceBuffer.size(); ++uTsIndex) {
      if (false == fdbCompSentFlags[uTsIndex][uCompIndex]) break;
    }  // for( ; uTsIndex < fdpTimesliceBuffer.size(); ++uTsIndex )

    if (fdpTimesliceBuffer.size() == uTsIndex) {
      --uTsIndex;
      if (!AddNewTsInBuffer()) return false;
    }  // if( fdpTimesliceBuffer.size() == uTsIndex )

    fles::StorableTimeslice component {static_cast<uint32_t>(fdpTimesliceBuffer[uTsIndex]->num_core_microslices()),
                                       fdpTimesliceBuffer[uTsIndex]->index()};
 
    for (uint32_t uComp = 0; uComp < fvvCompPerSysId[uCompIndex].size(); ++uComp) {
      uint32_t uNumMsInComp = fdpTimesliceBuffer[uTsIndex]->num_microslices(fvvCompPerSysId[uCompIndex][uComp]);
      component.append_component(uNumMsInComp);
 
      LOG(debug) << "Add components to TS for SysId " << std::hex << fSysId[uCompIndex] << std::dec << " TS "
                 << fdpTimesliceBuffer[uTsIndex]->index() << " Comp " << fvvCompPerSysId[uCompIndex][uComp];
 
      for (size_t m = 0; m < uNumMsInComp; ++m) {
        component.append_microslice(uComp, m,
                                    fdpTimesliceBuffer[uTsIndex]->descriptor(fvvCompPerSysId[uCompIndex][uComp], m),
                                    fdpTimesliceBuffer[uTsIndex]->content(fvvCompPerSysId[uCompIndex][uComp], m));
      }  // for( size_t m = 0; m < uNumMsInComp; ++m )
    }    // for( uint32_t uComp = 0; uComp < fvvCompPerSysId[ uCompIndex ].size(); ++uComp )
 
    LOG(debug) << "Prepared timeslice for SysId " << std::hex << fSysId[uCompIndex] << std::dec << " with "
               << component.num_components() << " components";
 
    if (!SendData(component)) return false;
 
    fdbCompSentFlags[uTsIndex][uCompIndex] = true;
  }  // if (0 < fvvCompPerSysId[uCompIndex].size())
 
  return true;
}
 
bool CbmMQTsSamplerRepReq::PrepareCompListPerBlock()
{
  if (false == fbListCompPerBlockReady) {
    if (!PrepareCompListPerSysId()) return false;

    for (auto itBlock = fvBlocksToSend.begin(); itBlock != fvBlocksToSend.end(); ++itBlock) {
      auto uBlockIdx = itBlock - fvBlocksToSend.begin();
 
      for (auto itSys = (itBlock->second).begin(); itSys != (itBlock->second).end(); ++itSys) {
        const vector<int>::const_iterator pos = std::find(fSysId.begin(), fSysId.end(), *itSys);
        if (fSysId.end() != pos) {
          const vector<int>::size_type idxSys = pos - fSysId.begin();

          for (uint32_t uComp = 0; uComp < fvvCompPerSysId[idxSys].size(); ++uComp) {
            fvvCompPerBlock[uBlockIdx].push_back(fvvCompPerSysId[idxSys][uComp]);
          }  // for( uint32_t uComp = 0; uComp < fvvCompPerSysId[ idxSys ].size(); ++uComp )
        }    // if (fSysId.end() != pos)
        else {
          LOG(error) << "Error when building the components list for block " << itBlock->first;
          LOG(error) << "Did not find 0x" << std::hex << *itSys << std::dec << " in the list of known systems";
          return false;
        }  // else of if (fSysId.end() != pos)
      }    // for( auto itSys = (itBlock->second).begin(); itSys != (itBlock->second).end(); ++itSys )
    }      // for( auto itBlock = fvBlocksToSend.begin(); itBlock != fvBlocksToSend.end(); ++itBlock)
 
    fbListCompPerBlockReady = true;
  }  // if( false == fbListCompPerBlockReady )
 
  return true;
}
 
bool CbmMQTsSamplerRepReq::CreateCombinedComponentsPerBlock(std::string sBlockName)
{
  if (!PrepareCompListPerBlock()) return false;

  for (auto itKnownBlock = fvBlocksToSend.begin(); itKnownBlock != fvBlocksToSend.end(); ++itKnownBlock) {
    if ((*itKnownBlock).first == sBlockName) {
      auto uBlockIdx = itKnownBlock - fvBlocksToSend.begin();

      uint32_t uTsIndex = 0;
      for (; uTsIndex < fdpTimesliceBuffer.size(); ++uTsIndex) {
        if (false == fdbCompSentFlags[uTsIndex][uBlockIdx]) break;
      }  // for( ; uTsIndex < fdpTimesliceBuffer.size(); ++uTsIndex )

      if (fdpTimesliceBuffer.size() == uTsIndex) {
        --uTsIndex;
        if (!AddNewTsInBuffer()) return false;
      }  // if( fdpTimesliceBuffer.size() == uTsIndex )

      fles::StorableTimeslice component {static_cast<uint32_t>(fdpTimesliceBuffer[uTsIndex]->num_core_microslices()),
                                         fdpTimesliceBuffer[uTsIndex]->index()};
 
      for (uint32_t uComp = 0; uComp < fvvCompPerBlock[uBlockIdx].size(); ++uComp) {
        uint32_t uNumMsInComp = fdpTimesliceBuffer[uTsIndex]->num_microslices(fvvCompPerBlock[uBlockIdx][uComp]);
        component.append_component(uNumMsInComp);
 
        LOG(debug) << "Add components to TS for Block " << sBlockName << " TS " << fdpTimesliceBuffer[uTsIndex]->index()
                   << " Comp " << fvvCompPerBlock[uBlockIdx][uComp];
 
        for (size_t m = 0; m < uNumMsInComp; ++m) {
          component.append_microslice(uComp, m,
                                      fdpTimesliceBuffer[uTsIndex]->descriptor(fvvCompPerBlock[uBlockIdx][uComp], m),
                                      fdpTimesliceBuffer[uTsIndex]->content(fvvCompPerBlock[uBlockIdx][uComp], m));
        }  // for( size_t m = 0; m < uNumMsInComp; ++m )
      }    // for( uint32_t uComp = 0; uComp < fvvCompPerSysId[ uCompIndex ].size(); ++uComp )
 
      LOG(debug) << "Prepared timeslice for Block " << sBlockName << " with " << component.num_components()
                 << " components";
 
      if (!SendData(component)) return false;
 
      fdbCompSentFlags[uTsIndex][uBlockIdx] = true;
      return true;
    }  // if( (*itKnownBlock).first == sBlockName )
  }    // for( auto itKnownBlock = fvBlocksToSend.begin(); itKnownBlock != fvBlocksToSend.end(); ++itKnownBlock)

  LOG(error) << "Requested block " << sBlockName << " not found in the list of known blocks";
  return false;
}
 
bool CbmMQTsSamplerRepReq::SendFirstTsIndex()
{
  // create the message with the first timeslice index
  std::string sIndex = FormatDecPrintout(fulFirstTsIndex);
  // serialize the vector and create the message
  std::stringstream oss;
  boost::archive::binary_oarchive oa(oss);
  oa << sIndex;
  std::string* strMsg = new std::string(oss.str());
 
  FairMQMessagePtr msg(NewMessage(
    const_cast<char*>(strMsg->c_str()),  // data
    strMsg->length(),                    // size
    [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
    strMsg));  // object that manages the data
 
  // in case of error or transfer interruption,
  // return false to go to IDLE state
  // successfull transfer will return number of bytes
  // transfered (can be 0 if sending an empty message).
  if (Send(msg, fsChannelNameTsRequest) < 0) {
    LOG(error) << "Problem sending reply with first TS index";
    return false;
  }
 
  fulMessageCounter++;
  LOG(debug) << "Send message " << fulMessageCounter << " with a size of " << msg->GetSize();
 
  return true;
}
bool CbmMQTsSamplerRepReq::SendData(const fles::StorableTimeslice& component)
{
  // serialize the timeslice and create the message
  std::stringstream oss;
  boost::archive::binary_oarchive oa(oss);
  oa << component;
  std::string* strMsg = new std::string(oss.str());
 
  FairMQMessagePtr msg(NewMessage(
    const_cast<char*>(strMsg->c_str()),  // data
    strMsg->length(),                    // size
    [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
    strMsg));  // object that manages the data
 
  // in case of error or transfer interruption,
  // return false to go to IDLE state
  // successfull transfer will return number of bytes
  // transfered (can be 0 if sending an empty message).
  if (Send(msg, fsChannelNameTsRequest) < 0) {
    LOG(error) << "Problem sending data";
    return false;
  }
 
  fulMessageCounter++;
  LOG(debug) << "Send message " << fulMessageCounter << " with a size of " << msg->GetSize();
 
  return true;
}
bool CbmMQTsSamplerRepReq::SendMissedTsIdx(std::vector<uint64_t> vIndices)
{
  // serialize the vector and create the message
  std::stringstream oss;
  boost::archive::binary_oarchive oa(oss);
  oa << vIndices;
  std::string* strMsg = new std::string(oss.str());
 
  FairMQMessagePtr msg(NewMessage(
    const_cast<char*>(strMsg->c_str()),  // data
    strMsg->length(),                    // size
    [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
    strMsg));  // object that manages the data
 
  // in case of error or transfer interruption,
  // return false to go to IDLE state
  // successfull transfer will return number of bytes
  // transfered (can be 0 if sending an empty message).
  LOG(debug) << "Send data to channel " << fsChannelNameMissedTs;
  if (Send(msg, fsChannelNameMissedTs) < 0) {
    LOG(error) << "Problem sending missed TS indices to channel " << fsChannelNameMissedTs;
    return false;
  }  // if( Send( msg, fsChannelNameMissedTs ) < 0 )
 
  return true;
}
bool CbmMQTsSamplerRepReq::SendCommand(std::string sCommand)
{
  // serialize the vector and create the message
  std::stringstream oss;
  boost::archive::binary_oarchive oa(oss);
  oa << sCommand;
  std::string* strMsg = new std::string(oss.str());
 
  FairMQMessagePtr msg(NewMessage(
    const_cast<char*>(strMsg->c_str()),  // data
    strMsg->length(),                    // size
    [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
    strMsg));  // object that manages the data
 
  //  FairMQMessagePtr msg( NewMessage( const_cast<char*>( sCommand.c_str() ), // data
  //                                    sCommand.length(), // size
  //                                    []( void* /*data*/, void* object ){ delete static_cast< std::string * >( object ); },
  //                                    &sCommand ) ); // object that manages the data
 
  // in case of error or transfer interruption,
  // return false to go to IDLE state
  // successfull transfer will return number of bytes
  // transfered (can be 0 if sending an empty message).
  LOG(debug) << "Send data to channel " << fsChannelNameCommands;
  if (Send(msg, fsChannelNameCommands) < 0) {
    LOG(error) << "Problem sending missed TS indices to channel " << fsChannelNameCommands;
    return false;
  }  // if( Send( msg, fsChannelNameMissedTs ) < 0 )
 
  return true;
}
bool CbmMQTsSamplerRepReq::SendHistoConfAndData()
{
  std::pair<uint32_t, uint32_t> pairHeader(fvpsHistosFolder.size(), fvpsCanvasConfig.size());
  FairMQMessagePtr messageHeader(NewMessage());
  //  Serialize<BoostSerializer<std::pair<uint32_t, uint32_t>>>(*messageHeader, pairHeader);
  BoostSerializer<std::pair<uint32_t, uint32_t>>().Serialize(*messageHeader, pairHeader);
 
  FairMQParts partsOut;
  partsOut.AddPart(std::move(messageHeader));
 
  for (UInt_t uHisto = 0; uHisto < fvpsHistosFolder.size(); ++uHisto) {
    FairMQMessagePtr messageHist(NewMessage());
    //    Serialize<BoostSerializer<std::pair<std::string, std::string>>>(*messageHist, fvpsHistosFolder[uHisto]);
    BoostSerializer<std::pair<std::string, std::string>>().Serialize(*messageHist, fvpsHistosFolder[uHisto]);
 
    partsOut.AddPart(std::move(messageHist));
  }  // for (UInt_t uHisto = 0; uHisto < fvpsHistosFolder.size(); ++uHisto)

  if (0 == fvpsHistosFolder.size()) {
    FairMQMessagePtr messageHist(NewMessage());
    partsOut.AddPart(std::move(messageHist));
  }
 
  for (UInt_t uCanv = 0; uCanv < fvpsCanvasConfig.size(); ++uCanv) {
    FairMQMessagePtr messageCan(NewMessage());
    //    Serialize<BoostSerializer<std::pair<std::string, std::string>>>(*messageCan, fvpsCanvasConfig[uCanv]);
    BoostSerializer<std::pair<std::string, std::string>>().Serialize(*messageCan, fvpsCanvasConfig[uCanv]);
 
    partsOut.AddPart(std::move(messageCan));
  }  // for (UInt_t uCanv = 0; uCanv < fvpsCanvasConfig.size(); ++uCanv)

  if (0 == fvpsCanvasConfig.size()) {
    FairMQMessagePtr messageHist(NewMessage());
    partsOut.AddPart(std::move(messageHist));
  }

  FairMQMessagePtr msgHistos(NewMessage());
  //  Serialize<RootSerializer>(*msgHistos, &fArrayHisto);
  RootSerializer().Serialize(*msgHistos, &fArrayHisto);
 
  partsOut.AddPart(std::move(msgHistos));

  if (Send(partsOut, fsChannelNameHistosInput) < 0) {
    LOG(error) << "CbmMQTsSamplerRepReq::SendHistoConfAndData => Problem sending data";
    return false;
  }  // if( Send( partsOut, fsChannelNameHistosInput ) < 0 )

  ResetHistograms();
 
  return true;
}
 
bool CbmMQTsSamplerRepReq::SendHistograms()
{
  FairMQMessagePtr message(NewMessage());
  //  Serialize<RootSerializer>(*message, &fArrayHisto);
  RootSerializer().Serialize(*message, &fArrayHisto);

  if (Send(message, fsChannelNameHistosInput) < 0) {
    LOG(error) << "Problem sending data";
    return false;
  }  // if( Send( message, fsChannelNameHistosInput ) < 0 )

  ResetHistograms();
 
  return true;
}
 
bool CbmMQTsSamplerRepReq::ResetHistograms()
{
  fhTsRate->Reset();
  fhTsSize->Reset();
  fhTsSizeEvo->Reset();
  fhTsMaxSizeEvo->Reset();
  fhMissedTS->Reset();
  fhMissedTSEvo->Reset();
 
  return true;
}
 
CbmMQTsSamplerRepReq::~CbmMQTsSamplerRepReq() {}
 
void CbmMQTsSamplerRepReq::CalcRuntime()
{
  std::chrono::duration<double> run_time = std::chrono::steady_clock::now() - fTime;
 
  LOG(info) << "Runtime: " << run_time.count();
  LOG(info) << "No more input data";
}