SourceEmulatorModel.hpp

Go to the documentation of this file.

 
#ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_MODELS_SOURCEEMULATORMODEL_HPP_
#define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_MODELS_SOURCEEMULATORMODEL_HPP_
 
#include "iomanager/IOManager.hpp"
#include "iomanager/Sender.hpp"
 
#include "logging/Logging.hpp"
 
#include "confmodel/DetectorStream.hpp"
#include "confmodel/GeoId.hpp"
 
#include "datahandlinglibs/DataHandlingIssues.hpp"
#include "datahandlinglibs/concepts/SourceEmulatorConcept.hpp"
#include "datahandlinglibs/utils/ErrorBitGenerator.hpp"
#include "datahandlinglibs/utils/FileSourceBuffer.hpp"
#include "datahandlinglibs/DataMoveCallbackRegistry.hpp"
#include "datahandlinglibs/utils/RateLimiter.hpp"
#include "utilities/ReusableThread.hpp"
 
#include "datahandlinglibs/opmon/datahandling_info.pb.h"
 
#include "unistd.h"
#include <chrono>
#include <functional>
#include <memory>
#include <random>
#include <string>
#include <utility>
#include <vector>
 
using dunedaq::datahandlinglibs::logging::TLVL_TAKE_NOTE;
using dunedaq::datahandlinglibs::logging::TLVL_WORK_STEPS;
 
namespace dunedaq {
namespace datahandlinglibs {
 
 
// Pattern generator class for creating different types of TP patterns
// AAA: at the moment the pattern generator is very simple: random source id and random channel
class SourceEmulatorPatternGenerator {
 
public:
  SourceEmulatorPatternGenerator() {
  };
  ~SourceEmulatorPatternGenerator() {};
 
  void generate(int, int);
 
  int get_channel_number() {
    if (m_index == m_channel.size()) m_index = 0;
    return m_channel[m_index++];
  }
 
private:
  std::vector<int> m_channel;
  int m_index = 0;
};
 
 
template<class ReadoutType>
class SourceEmulatorModel : public SourceEmulatorConcept
{
public:
  explicit SourceEmulatorModel(std::string name,
                               std::atomic<bool>& run_marker,
                               uint64_t time_tick_diff, // NOLINT(build/unsigned)
                               double dropout_rate,
                               double frame_error_rate,
                               double rate_khz,
                   uint16_t frames_per_tick=1)
    : m_run_marker(run_marker)
    , m_time_tick_diff(time_tick_diff)
    , m_dropout_rate(dropout_rate)
    , m_frame_error_rate(frame_error_rate)
    , m_packet_count{ 0 }
    , m_producer_thread(0)
    , m_name(name)
    , m_rate_khz(rate_khz)
    ,m_frames_per_tick(frames_per_tick)
  {}
 
  //void init(const appfwk::DAQModule::CommandData_t& /*args*/) {}
  void set_sender(const appmodel::DataMoveCallbackConf* conf);
 
  void conf(const confmodel::DetectorStream* stream_conf, const appmodel::StreamEmulationParameters* emu_conf);
  void scrap(const appfwk::DAQModule::CommandData_t& /*args*/)
  {
    m_file_source.reset();
    m_is_configured = false;
  }
  bool is_configured() override { return m_is_configured; }
 
  void start(const appfwk::DAQModule::CommandData_t& /*args*/);
  void stop(const appfwk::DAQModule::CommandData_t& /*args*/);
  //  void get_info(opmonlib::InfoCollector& ci, int /*level*/);
 
protected:
  // The data emulator function that the worker thread runs
  void run_produce();
  virtual void generate_opmon_data() override;
 
private:
  // Constuctor params
  std::atomic<bool>& m_run_marker;
 
  // CONFIGURATION
  uint32_t m_this_apa_number;  // NOLINT(build/unsigned)
  uint32_t m_this_link_number; // NOLINT(build/unsigned)
 
  uint64_t m_time_tick_diff; // NOLINT(build/unsigned)
  double m_dropout_rate;
  double m_frame_error_rate;
 
  // STATS
  std::atomic<int> m_packet_count{ 0 };
  std::atomic<int> m_packet_count_tot{ 0 };
 
  //sourceemulatorconfig::Conf m_cfg;
 
  bool m_sender_is_set = false;
  std::shared_ptr<std::function<void(ReadoutType&&)>> m_raw_data_callback;
  //using module_conf_t = dunedaq::datahandlinglibs::sourceemulatorconfig::Conf;
  //module_conf_t m_conf;
  //using link_conf_t = dunedaq::datahandlinglibs::sourceemulatorconfig::LinkConfiguration;
  //link_conf_t m_link_conf;
 
  std::unique_ptr<RateLimiter> m_rate_limiter;
  std::unique_ptr<FileSourceBuffer> m_file_source;
  ErrorBitGenerator m_error_bit_generator;
 
  utilities::ReusableThread m_producer_thread;
 
  std::string m_name;
  bool m_is_configured = false;
  double m_rate_khz;
  uint16_t m_frames_per_tick;
 
  std::vector<bool> m_dropouts; // Random population
  std::vector<bool> m_frame_errors;
 
  uint m_dropouts_length; // NOLINT(build/unsigned) Random population size
  uint m_frame_errors_length;
  daqdataformats::SourceID m_sourceid;
  int m_crateid;
  int m_slotid;
  int m_linkid;
 
  bool m_t0_now;
  // Pattern generator configs
  bool m_generate_periodic_adc_pattern;
  SourceEmulatorPatternGenerator m_pattern_generator;
  uint64_t m_pattern_generator_previous_ts;
  // Adding a hit every 9768 gives a TP rate of approx 100 Hz/wire using WIBEthernet
  uint32_t m_time_to_wait = 9768;
};
 
} // namespace datahandlinglibs
} // namespace dunedaq
 
// Declarations
#include "detail/SourceEmulatorModel.hxx"
 
#endif // DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_MODELS_SOURCEEMULATORMODEL_HPP_