CbmMQTsaMultiSampler.cxx

Go to the documentation of this file.

/* Copyright (C) 2017-2021 GSI Helmholtzzentrum fuer Schwerionenforschung, Darmstadt
   SPDX-License-Identifier: GPL-3.0-only
   Authors: Pierre-Alain Loizeau, Florian Uhlig [committer] */

 
 
#include "CbmMQTsaMultiSampler.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;
};
 
CbmMQTsaMultiSampler::CbmMQTsaMultiSampler()
  : FairMQDevice()
  , fMaxTimeslices(0)
  , fFileName("")
  , fDirName("")
  , fInputFileList()
  , fFileCounter(0)
  , fHost("")
  , fPort(0)
  , fHighWaterMark(1)
  , fTSCounter(0)
  , fMessageCounter(0)
  , fSource(nullptr)
  , fTime()
  , fLastPublishTime {std::chrono::system_clock::now()}
{
}
 
void CbmMQTsaMultiSampler::InitTask()
try {
  // Get the values from the command line options (via fConfig)
  fFileName                 = fConfig->GetValue<string>("filename");
  fDirName                  = fConfig->GetValue<string>("dirname");
  fHost                     = fConfig->GetValue<string>("flib-host");
  fPort                     = fConfig->GetValue<uint64_t>("flib-port");
  fHighWaterMark            = fConfig->GetValue<uint64_t>("high-water-mark");
  fMaxTimeslices            = fConfig->GetValue<uint64_t>("max-timeslices");
  fbNoSplitTs               = fConfig->GetValue<bool>("no-split-ts");
  fbSendTsPerSysId          = fConfig->GetValue<bool>("send-ts-per-sysid");
  fbSendTsPerChannel        = fConfig->GetValue<bool>("send-ts-per-channel");
  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");
  fsChannelNameHistosInput  = fConfig->GetValue<std::string>("ChNameIn");
  fsChannelNameHistosConfig = fConfig->GetValue<std::string>("ChNameHistCfg");
  fsChannelNameCanvasConfig = fConfig->GetValue<std::string>("ChNameCanvCfg");
 
  if (fbNoSplitTs) {
    if (fbSendTsPerSysId) {
      if (fbSendTsPerChannel) {
        LOG(warning) << "Both no-split-ts, send-ts-per-sysid and "
                        "send-ts-per-channel options used => "
                     << " second and third one will be ignored!!!!";
      }  // if( fbSendTsPerSysId )
      else
        LOG(warning) << "Both no-split-ts and send-ts-per-sysid options used => "
                     << " second one will be ignored!!!!";
    }  // if( fbSendTsPerSysId )
    else if (fbSendTsPerChannel) {
      LOG(warning) << "Both no-split-ts and send-ts-per-channel options used => "
                   << " second one will be ignored!!!!";
    }  // else if( fbSendTsPerSysId ) of if( fbSendTsPerSysId )
  }    // if( fbNoSplitTs )
  else if (fbSendTsPerSysId && fbSendTsPerChannel) {
    LOG(warning) << "Both send-ts-per-sysid and send-ts-per-channel options used => "
                 << " second one will be ignored!!!!";
  }  // else if( fbSendTsPerSysId && fbSendTsPerSysId ) of if( fbNoSplitTs )

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

    std::string sSysId  = vSysIdChanPairs[uPair].substr(0, sep);
    const size_t hexPos = sSysId.find("0x");
    int iSysId;
    if (string::npos == hexPos) iSysId = std::stoi(sSysId);
    else
      iSysId = std::stoi(sSysId.substr(hexPos + 2), nullptr, 16);

    std::string sChannelName = vSysIdChanPairs[uPair].substr(sep + 1);

    const vector<int>::const_iterator pos = std::find(fSysId.begin(), fSysId.end(), iSysId);
    if (fSysId.end() != pos) {
      const vector<std::string>::size_type idx = pos - fSysId.begin();
      fAllowedChannels[idx]                    = sChannelName;
    }  // if( fSysId.end() != pos )
    else {
      fSysId.push_back(iSysId);
      fAllowedChannels.push_back(sChannelName);
    }  // else of if( fSysId.end() != pos )
 
    LOG(info) << vSysIdChanPairs[uPair] << " " << iSysId << " " << sChannelName;
  }  // for( uint32_t uPair = 0; uPair < vSysIdChanPairs.size(); ++uPair )
 
  if (0 == fMaxTimeslices) fMaxTimeslices = UINT_MAX;
 
  // Check which input is defined
  // Posibilities
  // filename && ! dirname : single file
  // filename with wildcards && diranme : all files with filename regex in the directory
  // host && port : connect to the flim server
 
  bool isGoodInputCombi {false};
  if (0 != fFileName.size() && 0 == fDirName.size() && 0 == fHost.size() && 0 == fPort) {
    isGoodInputCombi = true;
    fInputFileList.push_back(fFileName);
  }
  else if (0 != fFileName.size() && 0 != fDirName.size() && 0 == fHost.size() && 0 == fPort) {
    isGoodInputCombi = true;
    fInputFileList.push_back(fFileName);
  }
  else if (0 == fFileName.size() && 0 == fDirName.size() && 0 != fHost.size() && 0 != fPort) {
    isGoodInputCombi = true;
    LOG(info) << "Host: " << fHost;
    LOG(info) << "Port: " << fPort;
  }
  else if (0 == fFileName.size() && 0 == fDirName.size() && 0 != fHost.size() && 0 == fPort) {
    isGoodInputCombi = true;
    LOG(info) << "Host string: " << fHost;
  }
  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: " << fMaxTimeslices;
 
  // Get the information about created channels from the device
  // Check if the defined channels from the topology (by name)
  // are in the list of channels which are possible/allowed
  // for the device
  // The idea is to check at initilization if the devices are
  // properly connected. For the time beeing this is done with a
  // nameing convention. It is not avoided that someone sends other
  // data on this channel.
  int noChannel = fChannels.size();
  LOG(info) << "Number of defined output channels: " << noChannel;
  for (auto const& entry : fChannels) {
    if (entry.first == fsChannelNameMissedTs || entry.first == fsChannelNameCommands
        || (0 < fuPublishFreqTs
            && (entry.first == fsChannelNameHistosInput || entry.first == fsChannelNameHistosConfig
                || entry.first == fsChannelNameCanvasConfig))) {
      continue;
    }  // if( entry.first == fsChannelNameMissedTs || entry.first == fsChannelNameCommands || histo channels name)
 
    LOG(info) << "Channel name: " << entry.first;
    if (!IsChannelNameAllowed(entry.first)) throw InitTaskError("Channel name does not match.");
  }
 
  for (auto const& value : fComponentsToSend) {
    LOG(info) << "Value : " << value;
    if (value > 1) {
      throw InitTaskError("Sending same data to more than one output channel "
                          "not implemented yet.");
    }
  }
 
 
  if (0 == fFileName.size() && 0 != fHost.size() && 0 != fPort) {
    // Don't add the protocol since this is done now in the TimesliceMultiSubscriber
    //std::string connector = "tcp://" + fHost + ":" + std::to_string(fPort);
    std::string connector = fHost + ":" + std::to_string(fPort);
    LOG(info) << "Open TSPublisher at " << connector;
    fSource = new fles::TimesliceMultiSubscriber(connector);
  }
  else if (0 == fFileName.size() && 0 != fHost.size()) {
    std::string connector = fHost;
    LOG(info) << "Open TSPublisher with host string: " << connector;
    fSource = new fles::TimesliceMultiSubscriber(connector, fHighWaterMark);
  }
  else {
    // Create a ";" separated string with all file names
    std::string fileList {""};
    for (const auto& obj : fInputFileList) {
      std::string fileName = obj;
      fileList += fileName;
      fileList += ";";
    }
    fileList.pop_back();  // Remove the last ;
    LOG(info) << "Input File String: " << fileList;
    fSource = new fles::TimesliceMultiInputArchive(fileList, fDirName);
    if (!fSource) { throw InitTaskError("Could open files from file list."); }
  }
 
  LOG(info) << "High-Water Mark: " << fHighWaterMark;
  LOG(info) << "Max. Timeslices: " << fMaxTimeslices;
  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 && fbSendTsPerSysId ) of if( fbNoSplitTs
  else if (fbSendTsPerChannel) {
    LOG(info) << "Sending components in separate TS per channel";
  }  // else if( fbSendTsPerSysId && fbSendTsPerSysId ) of if( fbNoSplitTs )
 
  fTime = std::chrono::steady_clock::now();
}
catch (InitTaskError& e) {
  LOG(error) << e.what();
  ChangeState(fair::mq::Transition::ErrorFound);
}
 
bool CbmMQTsaMultiSampler::IsChannelNameAllowed(std::string channelName)
{
  if (fbNoSplitTs) {
    fComponentsToSend[0]++;
    fChannelsToSend[0].push_back(channelName);
    return true;
  }  // if( fbNoSplitTs )
 
  bool bFoundMatch = false;
  //  for(auto const &entry : fAllowedChannels) {
  for (uint32_t idx = 0; idx < fAllowedChannels.size(); ++idx) {
    auto const& entry = fAllowedChannels[idx];
    LOG(info) << "Looking for name " << channelName << " in " << entry;
    std::size_t pos1 = channelName.find(entry);
    if (pos1 != std::string::npos) {
      /*
      const vector<std::string>::const_iterator pos =
         std::find(fAllowedChannels.begin(), fAllowedChannels.end(), entry);
      const vector<std::string>::size_type idx = pos-fAllowedChannels.begin();
*/
      LOG(info) << "Found " << entry << " in " << channelName;
      LOG(info) << "Channel name " << channelName << " found in list of allowed channel names at position " << idx
                << " (SysId 0x" << std::hex << fSysId[idx] << std::dec << ")";
      fComponentsToSend[idx]++;
      fChannelsToSend[idx].push_back(channelName);

      if (fbSendTsPerChannel) bFoundMatch = true;
      else
        return true;
    }  // if (pos1!=std::string::npos)
  }
  if (fbSendTsPerChannel && bFoundMatch) return true;
 
  LOG(info) << "Channel name " << channelName << " not found in list of allowed channel names.";
  LOG(error) << "Stop device.";
  return false;
}
 
bool CbmMQTsaMultiSampler::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;

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

  fhTsRate       = new TH1I("TsRate", "TS rate; t [s]", 1800, 0., 1800.);
  fhTsSize       = new TH1I("TsSize", "Size of TS; Size [MB]", 15000, 0., 15000.);
  fhTsSizeEvo    = new TProfile("TsSizeEvo", "Evolution of the TS Size; t [s]; Mean size [MB]", 1800, 0., 1800.);
  fhTsMaxSizeEvo = new TH1F("TsMaxSizeEvo", "Evolution of maximal TS Size; t [s]; Max size [MB]", 1800, 0., 1800.);
  fhMissedTS     = new TH1I("Missed_TS", "Missed TS", 2, -0.5, 1.5);
  fhMissedTSEvo  = new TProfile("Missed_TS_Evo", "Missed TS evolution; t [s]", 1800, 0., 1800.);

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

  Double_t w = 10;
  Double_t h = 10;
  fcSummary  = new TCanvas("cSampSummary", "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, "canvases"));

  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);

    FairMQMessagePtr messageHist(NewMessage());
    //    Serialize<BoostSerializer<std::pair<std::string, std::string>>>(*messageHist, psHistoConfig);
    BoostSerializer<std::pair<std::string, std::string>>().Serialize(*messageHist, psHistoConfig);

    if (Send(messageHist, fsChannelNameHistosConfig) < 0) {
      LOG(fatal) << "Problem sending histo config";
    }  // if( Send( messageHist, fsChannelNameHistosConfig ) < 0 )
 
    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);

    FairMQMessagePtr messageCan(NewMessage());
    //    Serialize<BoostSerializer<std::pair<std::string, std::string>>>(*messageCan, psCanvConfig);
    BoostSerializer<std::pair<std::string, std::string>>().Serialize(*messageCan, psCanvConfig);

    if (Send(messageCan, fsChannelNameCanvasConfig) < 0) {
      LOG(fatal) << "Problem sending canvas config";
    }  // if( Send( messageCan, fsChannelNameCanvasConfig ) < 0 )
 
    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 CbmMQTsaMultiSampler::ConditionalRun()
{
  if (0 < fuPublishFreqTs && 0 == fTSCounter) { InitHistograms(); }  // if( 0 < fuPublishFreqTs )

  if (0 == fTSCounter && nullptr != dynamic_cast<fles::TimesliceMultiSubscriber*>(fSource)) {
    dynamic_cast<fles::TimesliceMultiSubscriber*>(fSource)->InitTimesliceSubscriber();
  }  // if( 0 == fTSCounter && nullptr != dynamic_cast< fles::TimesliceMultiSubscriber >(fSource) )
 
  auto timeslice = fSource->get();
  if (timeslice) {
    if (fTSCounter < fMaxTimeslices) {
      fTSCounter++;
 
      const fles::Timeslice& ts = *timeslice;
      uint64_t uTsIndex         = ts.index();
 
      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 != (fuPrevTsIndex + 1)) && !(0 == fuPrevTsIndex && 0 == uTsIndex)) {
        LOG(info) << "Missed Timeslices. Old TS Index was " << fuPrevTsIndex << " New TS Index is " << uTsIndex
                  << " diff is " << uTsIndex - fuPrevTsIndex << " Missing are " << uTsIndex - fuPrevTsIndex - 1;
 
        if ("" != fsChannelNameMissedTs) {
          std::vector<uint64_t> vulMissedIndices;
          for (uint64_t ulMiss = fuPrevTsIndex + 1; ulMiss < uTsIndex; ++ulMiss) {
            vulMissedIndices.emplace_back(ulMiss);
          }  // for( uint64_t ulMiss = fuPrevTsIndex + 1; ulMiss < uTsIndex; ++ulMiss )
          if (!SendMissedTsIdx(vulMissedIndices)) {
            if ("" != fsChannelNameCommands) {
              std::this_thread::sleep_for(std::chrono::milliseconds(1000));
              SendCommand("STOP");
            }  // if( "" != fsChannelNameCommands )
 
            return false;
          }  // if( !SendMissedTsIdx( vulMissedIndices ) )
        }    // if( "" != fsChannelNameMissedTs )
 
        if (0 < fuPublishFreqTs) {
          fhMissedTS->Fill(1, uTsIndex - fuPrevTsIndex - 1);
          fhMissedTSEvo->Fill(fdTimeToStart, 1, uTsIndex - fuPrevTsIndex - 1);
        }  // if( 0 < fuPublishFreqTs )
 
      }  // if( ( uTsIndex != ( fuPrevTsIndex + 1 ) ) && !( 0 == fuPrevTsIndex && 0 == uTsIndex ) )
 
      if (0 < fuPublishFreqTs) {
        fhMissedTS->Fill(0);
        fhMissedTSEvo->Fill(fdTimeToStart, 0, 1);
      }  // else if( 0 < fuPublishFreqTs )
 
      fuPrevTsIndex = uTsIndex;
 
      if (fTSCounter % 10000 == 0) { LOG(info) << "Received TS " << fTSCounter << " with index " << uTsIndex; }
 
      LOG(debug) << "Found " << ts.num_components() << " different components in timeslice";
 
 
      //      CheckTimeslice(ts);
 
      if (fbNoSplitTs) {
        if (!CreateAndSendFullTs(ts)) {
          if ("" != fsChannelNameCommands) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            SendCommand("STOP");
          }  // if( "" != fsChannelNameCommands )
 
          return false;
        }  // if( !CreateAndSendFullTs( ts ) )
      }    // if( fbNoSplitTs )
      else if (fbSendTsPerSysId) {
        if (!CreateAndCombineComponentsPerSysId(ts)) {
          if ("" != fsChannelNameCommands) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            SendCommand("STOP");
          }  // if( "" != fsChannelNameCommands )
 
          return false;
        }  // if( !CreateAndCombineComponentsPerSysId( ts ) )
      }    // else if( fbSendTsPerSysId ) of  if( fbNoSplitTs )
      else if (fbSendTsPerChannel) {
        if (!CreateAndCombineComponentsPerChannel(ts)) {
          if ("" != fsChannelNameCommands) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            SendCommand("STOP");
          }  // if( "" != fsChannelNameCommands )
 
          return false;
        }  // if( !CreateAndCombineComponentsPerChannel( ts ) )
      }    // else if( fbSendTsPerChannel ) of  if( fbSendTsPerSysId )
      else {
        for (unsigned int nrComp = 0; nrComp < ts.num_components(); ++nrComp) {
          if (!CreateAndSendComponent(ts, nrComp)) {
            if ("" != fsChannelNameCommands) {
              std::this_thread::sleep_for(std::chrono::milliseconds(1000));
              SendCommand("STOP");
            }  // if( "" != fsChannelNameCommands )
 
            return false;
          }  // if( !CreateAndSendComponent(ts, nrComp) )
        }    // for (unsigned int nrComp = 0; nrComp < ts.num_components(); ++nrComp)
      }      // else of if( fbSendTsPerSysId )
 
      if (0 < fuPublishFreqTs) {
        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 == fTSCounter % fuPublishFreqTs && fdMinPublishTime < elapsedSeconds.count())) {
          SendHistograms();
          fLastPublishTime = std::chrono::system_clock::now();
        }  // if( ( fdMaxPublishTime < elapsedSeconds.count() ) || ( 0 == fTSCounter % fuPublishFreqTs && fdMinPublishTime < elapsedSeconds.count() ) )
      }    // if( 0 < fuPublishFreqTs )
 
      return true;
    }  // if (fTSCounter < fMaxTimeslices)
    else {
      CalcRuntime();

      if ("" != fsChannelNameCommands) {
        std::this_thread::sleep_for(std::chrono::seconds(10));
        std::string sCmd = "EOF ";
        sCmd += FormatDecPrintout(fuPrevTsIndex);
        sCmd += " ";
        sCmd += FormatDecPrintout(fTSCounter);
        SendCommand(sCmd);
      }  // if( "" != fsChannelNameCommands )
 
      return false;
    }  // else of if (fTSCounter < fMaxTimeslices)
  }    // if (timeslice)
  else {
    CalcRuntime();

    if ("" != fsChannelNameCommands) {
      std::this_thread::sleep_for(std::chrono::seconds(10));
      std::string sCmd = "EOF ";
      sCmd += FormatDecPrintout(fuPrevTsIndex);
      sCmd += " ";
      sCmd += FormatDecPrintout(fTSCounter);
      SendCommand(sCmd);
    }  // if( "" != fsChannelNameCommands )
 
    return false;
  }  // else of if (timeslice)
}
 
bool CbmMQTsaMultiSampler::CreateAndSendComponent(const fles::Timeslice& ts, int nrComp)
{
 
  // Check if component has to be send. If the corresponding channel
  // is connected create the new timeslice and send it to the
  // correct channel
 
  LOG(debug) << "SysID: " << static_cast<int>(ts.descriptor(nrComp, 0).sys_id);
  const vector<int>::const_iterator pos =
    std::find(fSysId.begin(), fSysId.end(), static_cast<int>(ts.descriptor(nrComp, 0).sys_id));
  if (pos != fSysId.end()) {
    const vector<std::string>::size_type idx = pos - fSysId.begin();
    if (fComponentsToSend[idx] > 0) {
      LOG(debug) << "Create timeslice component for link " << nrComp;
 
      fles::StorableTimeslice component {static_cast<uint32_t>(ts.num_core_microslices()), ts.index()};
      component.append_component(ts.num_microslices(0));
 
      for (size_t m = 0; m < ts.num_microslices(nrComp); ++m) {
        component.append_microslice(0, m, ts.descriptor(nrComp, m), ts.content(nrComp, m));
      }
      /*
      LOG(info) << "Number of core microslices before: " << ts.num_core_microslices();
      LOG(info) << "Number of core microslices after : " << component.num_core_microslices();
      LOG(info) << "Number of microslices: " << component.num_microslices(0);
*/
      if (!SendData(component, idx)) return false;
      return true;
    }
  }
  return true;
}
 
bool CbmMQTsaMultiSampler::CreateAndCombineComponentsPerSysId(const fles::Timeslice& ts)
{
  if (false == fbListCompPerSysIdReady) {
    for (uint32_t uCompIdx = 0; uCompIdx < ts.num_components(); ++uCompIdx) {
      uint16_t usSysId = ts.descriptor(uCompIdx, 0).sys_id;
 
      const vector<int>::const_iterator pos = std::find(fSysId.begin(), fSysId.end(), usSysId);
      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 uNrComp = 0; uNrComp < ts.num_components(); ++uNrComp )
 
    for (uint32_t uSysIdx = 0; uSysIdx < fComponentsToSend.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 )

  for (uint32_t uSysIdx = 0; uSysIdx < fComponentsToSend.size(); ++uSysIdx) {
    if (0 < fComponentsToSend[uSysIdx]) {
      LOG(debug) << "Create timeslice with components for SysId " << std::hex << fSysId[uSysIdx] << std::dec;
 
      if (0 < fvvCompPerSysId[uSysIdx].size()) {
        fles::StorableTimeslice component {static_cast<uint32_t>(ts.num_core_microslices()), ts.index()};
 
        for (uint32_t uComp = 0; uComp < fvvCompPerSysId[uSysIdx].size(); ++uComp) {
          uint32_t uNumMsInComp = ts.num_microslices(fvvCompPerSysId[uSysIdx][uComp]);
          component.append_component(uNumMsInComp);
 
          LOG(debug) << "Add components to TS for SysId " << std::hex << fSysId[uSysIdx] << std::dec << " TS "
                     << ts.index() << " Comp " << fvvCompPerSysId[uSysIdx][uComp];
 
          for (size_t m = 0; m < uNumMsInComp; ++m) {
            component.append_microslice(uComp, m, ts.descriptor(fvvCompPerSysId[uSysIdx][uComp], m),
                                        ts.content(fvvCompPerSysId[uSysIdx][uComp], m));
          }  // for( size_t m = 0; m < uNumMsInComp; ++m )
        }    // for( uint32_t uComp = 0; uComp < fvvCompPerSysId[ uSysIdx ].size(); ++uComp )
 
        LOG(debug) << "Prepared timeslice for SysId " << std::hex << fSysId[uSysIdx] << std::dec << " with "
                   << component.num_components() << " components";
 
        if (!SendData(component, uSysIdx)) return false;
      }  // if( 0 < fvvCompPerSysId[ uSysIdx ].size() )
    }    // if( 0 < fComponentsToSend[ uSysIdx ] )
  }      // for( uSysIdx = 0; uSysIdx < fComponentsToSend.size(); ++uSysIdx )
 
  return true;
}
 
bool CbmMQTsaMultiSampler::CreateAndCombineComponentsPerChannel(const fles::Timeslice& ts)
{

  if (false == fbListCompPerChannelReady) {
    for (uint32_t uSysIdx = 0; uSysIdx < fComponentsToSend.size(); ++uSysIdx) {
      if (0 < fComponentsToSend[uSysIdx]) {
        for (uint32_t uChan = 0; uChan < fChannelsToSend[uSysIdx].size(); ++uChan) {
          const vector<std::string>::const_iterator pos =
            std::find(fvChannelsToSend.begin(), fvChannelsToSend.end(), fChannelsToSend[uSysIdx][uChan]);
          if (fvChannelsToSend.end() == pos) {
            fvChannelsToSend.push_back(fChannelsToSend[uSysIdx][uChan]);
            fvvCompPerChannel.push_back(std::vector<uint32_t>());
          }
        }  // for( uChan = 0; uChan < fChannelsToSend[ uSysIdx ].size(); ++ uChan )
      }    // if( 0 < fComponentsToSend[ uSysIdx ] )
    }      // for( uint32_t uSysIdx = 0; uSysIdx < fComponentsToSend.size(); ++uSysIdx )

    fvvCompPerChannel.resize(fvChannelsToSend.size());

    for (uint32_t uCompIdx = 0; uCompIdx < ts.num_components(); ++uCompIdx) {
      uint16_t usSysId = ts.descriptor(uCompIdx, 0).sys_id;
 
      const vector<int>::const_iterator pos = std::find(fSysId.begin(), fSysId.end(), usSysId);
      if (fSysId.end() != pos) {
        const vector<std::string>::size_type idxSys = pos - fSysId.begin();
 
        if (0 < fComponentsToSend[idxSys]) {
          for (uint32_t uChan = 0; uChan < fChannelsToSend[idxSys].size(); ++uChan) {
            const vector<std::string>::const_iterator posCh =
              std::find(fvChannelsToSend.begin(), fvChannelsToSend.end(), fChannelsToSend[idxSys][uChan]);
            if (fvChannelsToSend.end() != posCh) {
              const vector<std::string>::size_type idxChan = posCh - fvChannelsToSend.begin();
              fvvCompPerChannel[idxChan].push_back(uCompIdx);
            }  // if( fvChannelsToSend.end() != posCh )
          }    // for( uChan = 0; uChan < fChannelsToSend[ idxSys ].size(); ++ uChan )
        }      // if( 0 < fComponentsToSend[ uSysIdx ] )
      }        // if( fSysId.end() != pos )
    }          // for( uint32_t uNrComp = 0; uNrComp < ts.num_components(); ++uNrComp )
 
    for (uint32_t uChanIdx = 0; uChanIdx < fvChannelsToSend.size(); ++uChanIdx) {
      std::stringstream ss;
      ss << "Found " << std::setw(2) << fvvCompPerChannel[uChanIdx].size() << " components for channel "
         << fvChannelsToSend[uChanIdx] << " :";
 
      for (uint32_t uComp = 0; uComp < fvvCompPerChannel[uChanIdx].size(); ++uComp) {
        ss << " " << std::setw(3) << fvvCompPerChannel[uChanIdx][uComp];
      }  // for( uint32_t uComp = 0; uComp < fvvCompPerChannel[ uChanIdx ].size(); ++uComp )
 
      LOG(info) << ss.str();
    }  // for( uint32_t uChanIdx = 0; uChanIdx < fvChannelsToSend.size(); ++uChanIdx )
 
    fbListCompPerChannelReady = true;
  }  // if( false == fbListCompPerSysIdReady )


  for (uint32_t uChanIdx = 0; uChanIdx < fvChannelsToSend.size(); ++uChanIdx) {
    LOG(debug) << "Create timeslice with components for channel " << fvChannelsToSend[uChanIdx];
 
    if (0 < fvvCompPerChannel[uChanIdx].size()) {
      fles::StorableTimeslice component {static_cast<uint32_t>(ts.num_core_microslices()), ts.index()};
 
      for (uint32_t uComp = 0; uComp < fvvCompPerChannel[uChanIdx].size(); ++uComp) {
        uint32_t uNumMsInComp = ts.num_microslices(fvvCompPerChannel[uChanIdx][uComp]);
        component.append_component(uNumMsInComp);
 
        LOG(debug) << "Add components to TS for SysId " << std::hex
                   << static_cast<uint16_t>(ts.descriptor(fvvCompPerChannel[uChanIdx][uComp], 0).sys_id) << std::dec
                   << " TS " << ts.index() << " Comp " << fvvCompPerChannel[uChanIdx][uComp];
 
        for (size_t m = 0; m < uNumMsInComp; ++m) {
          component.append_microslice(uComp, m, ts.descriptor(fvvCompPerChannel[uChanIdx][uComp], m),
                                      ts.content(fvvCompPerChannel[uChanIdx][uComp], m));
        }  // for( size_t m = 0; m < uNumMsInComp; ++m )
      }    // for( uint32_t uComp = 0; uComp < fvvCompPerChannel[ uChanIdx ].size(); ++uComp )
 
      LOG(debug) << "Prepared timeslice for channel " << fvChannelsToSend[uChanIdx] << " with "
                 << component.num_components() << " components";
 
      if (!SendData(component, fvChannelsToSend[uChanIdx])) return false;
    }  // if( 0 < fvvCompPerSysId[ uSysIdx ].size() )
  }    // for( uChanIdx = 0; uChanIdx < fvChannelsToSend.size(); ++uChanIdx )
 
  return true;
}
 
bool CbmMQTsaMultiSampler::CreateAndSendFullTs(const fles::Timeslice& ts)
{
  for (uint32_t uChanIdx = 0; uChanIdx < fChannelsToSend.size(); ++uChanIdx) {
    if (0 < fComponentsToSend[uChanIdx]) {
      LOG(debug) << "Copy timeslice component for channel " << fChannelsToSend[uChanIdx][0];
 
      fles::StorableTimeslice fullTs {ts};
      if (!SendData(fullTs, uChanIdx)) return false;
    }  // if( 0 < fComponentsToSend[ uChanIdx ] )
  }    // for( uint32_t uChanIdx = 0; uChanIdx < fChannelsToSend.size(); ++uChanIdx )
  return true;
}
 
bool CbmMQTsaMultiSampler::SendData(const fles::StorableTimeslice& component, int idx)
{
  // 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
 
  // TODO: Implement sending same data to more than one channel
  // Need to create new message (copy message??)
  if (fComponentsToSend[idx] > 1) { LOG(info) << "Need to copy FairMessage"; }
 
  // 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 " << fChannelsToSend[idx][0];
  if (Send(msg, fChannelsToSend[idx][0]) < 0) {
    LOG(error) << "Problem sending data";
    return false;
  }
 
  fMessageCounter++;
  LOG(debug) << "Send message " << fMessageCounter << " with a size of " << msg->GetSize();
 
  return true;
}
 
bool CbmMQTsaMultiSampler::SendData(const fles::StorableTimeslice& component, std::string sChannel)
{
  // 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).
  LOG(debug) << "Send data to channel " << sChannel;
  if (Send(msg, sChannel) < 0) {
    LOG(error) << "Problem sending data";
    return false;
  }
 
  fMessageCounter++;
  LOG(debug) << "Send message " << fMessageCounter << " with a size of " << msg->GetSize();
 
  return true;
}
bool CbmMQTsaMultiSampler::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 CbmMQTsaMultiSampler::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 CbmMQTsaMultiSampler::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 CbmMQTsaMultiSampler::ResetHistograms()
{
  fhTsRate->Reset();
  fhTsSize->Reset();
  fhTsSizeEvo->Reset();
  fhTsMaxSizeEvo->Reset();
  fhMissedTS->Reset();
  fhMissedTSEvo->Reset();
 
  return true;
}
 
CbmMQTsaMultiSampler::~CbmMQTsaMultiSampler() {}
 
void CbmMQTsaMultiSampler::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";
}
 
 
void CbmMQTsaMultiSampler::PrintMicroSliceDescriptor(const fles::MicrosliceDescriptor& mdsc)
{
  LOG(info) << "Header ID: Ox" << std::hex << static_cast<int>(mdsc.hdr_id) << std::dec;
  LOG(info) << "Header version: Ox" << std::hex << static_cast<int>(mdsc.hdr_ver) << std::dec;
  LOG(info) << "Equipement ID: " << mdsc.eq_id;
  LOG(info) << "Flags: " << mdsc.flags;
  LOG(info) << "Sys ID: Ox" << std::hex << static_cast<int>(mdsc.sys_id) << std::dec;
  LOG(info) << "Sys version: Ox" << std::hex << static_cast<int>(mdsc.sys_ver) << std::dec;
  LOG(info) << "Microslice Idx: " << mdsc.idx;
  LOG(info) << "Checksum: " << mdsc.crc;
  LOG(info) << "Size: " << mdsc.size;
  LOG(info) << "Offset: " << mdsc.offset;
}
 
bool CbmMQTsaMultiSampler::CheckTimeslice(const fles::Timeslice& ts)
{
  if (0 == ts.num_components()) {
    LOG(error) << "No Component in TS " << ts.index();
    return 1;
  }
  LOG(info) << "Found " << ts.num_components() << " different components in timeslice";
 
  for (size_t c = 0; c < ts.num_components(); ++c) {
    LOG(info) << "Found " << ts.num_microslices(c) << " microslices in component " << c;
    LOG(info) << "Component " << c << " has a size of " << ts.size_component(c) << " bytes";
    LOG(info) << "Component " << c << " has the system id 0x" << std::hex
              << static_cast<int>(ts.descriptor(c, 0).sys_id) << std::dec;
    LOG(info) << "Component " << c << " has the system id 0x" << static_cast<int>(ts.descriptor(c, 0).sys_id);
 
    /*
    for (size_t m = 0; m < ts.num_microslices(c); ++m) {
      PrintMicroSliceDescriptor(ts.descriptor(c,m));
    }
*/
  }
 
  return true;
}